diff --git a/.fernignore b/.fernignore index ae0c0bb..94457f9 100644 --- a/.fernignore +++ b/.fernignore @@ -2,3 +2,6 @@ README.md assets/ + +src/webflow/client.py +src/webflow/oauth.py diff --git a/README.md b/README.md index 8a54076..4a11228 100644 --- a/README.md +++ b/README.md @@ -26,7 +26,11 @@ Simply import `Webflow` and start making calls to our API. ```python from webflow.client import Webflow -client = Webflow(access_token="YOUR_ACCESS_TOKEN") +client = Webflow( + client_id="YOUR_CLIENT_ID", + client_secret="YOUR_CLIENT_SECRET", + code="YOUR_AUTHORIZATION_CODE" +) site = client.sites.get("site-id") ``` @@ -38,7 +42,9 @@ calls to our API. from webflow.client import AsyncWebflow client = AsyncWebflow( - access_token="YOUR_ACCESS_TOKEN", + client_id="YOUR_CLIENT_ID", + client_secret="YOUR_CLIENT_SECRET", + code="YOUR_AUTHORIZATION_CODE" ) async def main() -> None: @@ -48,6 +54,58 @@ async def main() -> None: asyncio.run(main()) ``` +## OAuth + +To implement OAuth, you'll need a registred Webflow App. + +### Step 1: Authorize URL + +The first step in OAuth is to generate an authorization url. Use this URL +to fetch your authorization code. See the [docs](https://docs.developers.webflow.com/v1.0.0/docs/oauth#user-authorization +for more details. + +```python +from webflow.oauth import authorize_url +from webflow import OauthScope + +url = webflow.authorize_url({ + client_id = "[CLIENT ID]", + scope = OauthScope.ReadUsers, # or [OauthScope.ReadUsers, OauthScope.WriteUsers] + state = "1234567890", # optional + redirect_uri = "https://my.server.com/oauth/callback", # optional +}); + +print(url) +``` + +### Step 2: Instantiate the client +Pass in your `client_id`, `client_secret`, `authorization_code` when instantiating +the client. Our SDK handles generating an access token and passing that to every endpoint. + +```python +from webflow.client import Webflow + +client = Webflow( + client_id="YOUR_CLIENT_ID", + client_secret="YOUR_CLIENT_SECRET", + code="YOUR_AUTHORIZATION_CODE", + redirect_uri = "https://my.server.com/oauth/callback", # optional +) +``` + +If you want to generate an access token yourself, simply import the +`get_access_token` function. + +```python +from webflow.oauth import get_access_token + +access_token = get_access_token( + client_id="YOUR_CLIENT_ID", + client_secret="YOUR_CLIENT_SECRET", + code="YOUR_AUTHORIZATION_CODE" + ) +``` + ## Webflow Module All of the models are nested within the Webflow module. Let Intellisense guide you! diff --git a/src/webflow/__init__.py b/src/webflow/__init__.py index 5835070..dc9642b 100644 --- a/src/webflow/__init__.py +++ b/src/webflow/__init__.py @@ -50,6 +50,7 @@ ListCustomCodeBlocks, MissingScopes, NoDomains, + OauthScope, Order, OrderAddress, OrderAddressJapanType, @@ -214,6 +215,7 @@ "MissingScopes", "NoDomains", "NotFoundError", + "OauthScope", "Order", "OrderAddress", "OrderAddressJapanType", diff --git a/src/webflow/client.py b/src/webflow/client.py index 08638bc..63d4dbf 100644 --- a/src/webflow/client.py +++ b/src/webflow/client.py @@ -6,6 +6,7 @@ from .core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from .environment import WebflowEnvironment +from .oauth import get_access_token from .resources.access_groups.client import AccessGroupsClient, AsyncAccessGroupsClient from .resources.assets.client import AssetsClient, AsyncAssetsClient from .resources.collections.client import AsyncCollectionsClient, CollectionsClient @@ -26,15 +27,22 @@ class Webflow: def __init__( self, *, - base_url: typing.Optional[str] = None, + client_id: str, + client_secret: str, + code: str, + redirect_uri: typing.Optional[str] = None, environment: WebflowEnvironment = WebflowEnvironment.DEFAULT, - access_token: typing.Union[str, typing.Callable[[], str]], timeout: typing.Optional[float] = 60, httpx_client: typing.Optional[httpx.Client] = None ): + self._token = get_access_token( + client_id=client_id, + client_secret=client_secret, + code=code, + redirect_uri=redirect_uri) self._client_wrapper = SyncClientWrapper( - base_url=_get_base_url(base_url=base_url, environment=environment), - access_token=access_token, + base_url=_get_base_url(base_url=None, environment=environment), + access_token=self._token, httpx_client=httpx.Client(timeout=timeout) if httpx_client is None else httpx_client, ) self.token = TokenClient(client_wrapper=self._client_wrapper) @@ -57,15 +65,22 @@ class AsyncWebflow: def __init__( self, *, - base_url: typing.Optional[str] = None, + client_id: str, + client_secret: str, + code: str, + redirect_uri: typing.Optional[str] = None, environment: WebflowEnvironment = WebflowEnvironment.DEFAULT, - access_token: typing.Union[str, typing.Callable[[], str]], timeout: typing.Optional[float] = 60, httpx_client: typing.Optional[httpx.AsyncClient] = None ): + self._token = get_access_token( + client_id=client_id, + client_secret=client_secret, + code=code, + redirect_uri=redirect_uri) self._client_wrapper = AsyncClientWrapper( - base_url=_get_base_url(base_url=base_url, environment=environment), - access_token=access_token, + base_url=_get_base_url(base_url=None, environment=environment), + access_token=self._token, httpx_client=httpx.AsyncClient(timeout=timeout) if httpx_client is None else httpx_client, ) self.token = AsyncTokenClient(client_wrapper=self._client_wrapper) diff --git a/src/webflow/oauth.py b/src/webflow/oauth.py new file mode 100644 index 0000000..4de369b --- /dev/null +++ b/src/webflow/oauth.py @@ -0,0 +1,114 @@ + +import typing +import httpx +import urllib.parse +from json.decoder import JSONDecodeError + +from .core.api_error import ApiError +from .core.jsonable_encoder import jsonable_encoder +from .environment import WebflowEnvironment +from .types import OauthScope + +try: + import pydantic.v1 as pydantic # type: ignore +except ImportError: + import pydantic # type: ignore + +# this is used as the default value for optional parameters +OMIT = typing.cast(typing.Any, ...) + + +def authorize_url( + *, + client_id: str, + state: typing.Optional[str] = OMIT, + redirect_uri: typing.Optional[str] = OMIT, + scope: typing.Optional[typing.Union[OauthScope, typing.List[OauthScope]]] = OMIT, +) -> str: + """ + Get the URL to authorize a user + + Parameters: + - client_id: str. The OAuth client ID + + - state: typing.Optional[str]. The state. + + - redirect_uri: typing.Optional[str]. The redirect URI. + + - scope: typing.Optional[typing.Union[OauthScope, typing.List[OauthScope]]]. + OAuth Scopes. + --- + from webflow.oauth import authorize_url + from webflow import OauthScope + + url = authorize_url( + client_id = "", + redirect_uri = "https://my.server.com/oauth/callback", + scopes = [OauthScope.ReadSites, OauthScope.WriteItems", OauthScope.ReadUsers], + ) + """ + params: typing.Dict[str, typing.Any] = { + "client_id": client_id, + "response_type": "code", + } + if state is not OMIT: + params["state"] = state + if redirect_uri is not OMIT: + params["redirect_uri"] = redirect_uri + if scope is not OMIT and isinstance(scope, str): + params["scope"] = scope.value + elif scope is not OMIT: + params["scope"] = ", ".join([s.value for s in scope]) # type: ignore + return f"https://webflow.com/oauth/authorize?{urllib.parse.urlencode(params)}" + + +def get_access_token( + *, + client_id: str, + client_secret: str, + code: str, + redirect_uri: typing.Optional[str] = OMIT, +) -> str: + """ + Get the URL to authorize a user + + Parameters: + - client_id: str. The OAuth client ID + + - client_secret: str. The OAuth client secret + + - code: str. The OAuth code + + - redirect_uri: typing.Optional[str]. The redirect URI. + --- + from webflow.oauth import get_access_token + + token = get_access_token( + client_id = "", + client_secret = "", + code= "" + redirect_uri = "https://my.server.com/oauth/callback", + ) + """ + request: typing.Dict[str, typing.Any] = { + "client_id": client_id, + "client_secret": client_secret, + "code": code, + "grant_type": "authorization_code", + } + if redirect_uri is not OMIT: + request["redirect_uri"] = redirect_uri + response = httpx.request( + "POST", + "https://api.webflow.com/oauth/access_token", + json=jsonable_encoder(request), + timeout=60, + ) + if 200 <= response.status_code < 300: + _response_json = response.json() + return _response_json["access_token"] + try: + raise ApiError(status_code=response.status_code, body=response.json()) + except JSONDecodeError: + raise ApiError(status_code=response.status_code, body=response.text) + diff --git a/src/webflow/types/__init__.py b/src/webflow/types/__init__.py index cb6b440..7282f1c 100644 --- a/src/webflow/types/__init__.py +++ b/src/webflow/types/__init__.py @@ -49,6 +49,7 @@ from .list_custom_code_blocks import ListCustomCodeBlocks from .missing_scopes import MissingScopes from .no_domains import NoDomains +from .oauth_scope import OauthScope from .order import Order from .order_address import OrderAddress from .order_address_japan_type import OrderAddressJapanType @@ -173,6 +174,7 @@ "ListCustomCodeBlocks", "MissingScopes", "NoDomains", + "OauthScope", "Order", "OrderAddress", "OrderAddressJapanType", diff --git a/src/webflow/types/oauth_scope.py b/src/webflow/types/oauth_scope.py new file mode 100644 index 0000000..4299b79 --- /dev/null +++ b/src/webflow/types/oauth_scope.py @@ -0,0 +1,100 @@ +# This file was auto-generated by Fern from our API Definition. + +import enum +import typing + +T_Result = typing.TypeVar("T_Result") + + +class OauthScope(str, enum.Enum): + AUTHORIZED_USER_READ = "authorized_user:read" + """ + read details about the authorized user + """ + + READ_PAGES = "read:pages" + """ + read pages on the site + """ + + SITES_READ = "sites:read" + """ + read sites on the site + """ + + SITES_WRITE = "sites:write" + """ + modify pages on the site + """ + + CUSTOM_CODE_READ = "custom_code:read" + """ + read custom code on the site + """ + + CUSTOM_CODE_WRITE = "custom_code:write" + """ + modify custom code on the site + """ + + CUSTOM_CODE_DELETE = "custom_code:delete" + """ + delete custom code on the site + """ + + USERS_READ = "users:read" + """ + read users on the site + """ + + USERS_WRITE = "users:write" + """ + modify users on the site + """ + + ECOMMERCE_READ = "ecommerce:read" + """ + read ecommerce data + """ + + ECOMMERCE_WRITE = "ecommerce:write" + """ + edit ecommerce data + """ + + def visit( + self, + authorized_user_read: typing.Callable[[], T_Result], + read_pages: typing.Callable[[], T_Result], + sites_read: typing.Callable[[], T_Result], + sites_write: typing.Callable[[], T_Result], + custom_code_read: typing.Callable[[], T_Result], + custom_code_write: typing.Callable[[], T_Result], + custom_code_delete: typing.Callable[[], T_Result], + users_read: typing.Callable[[], T_Result], + users_write: typing.Callable[[], T_Result], + ecommerce_read: typing.Callable[[], T_Result], + ecommerce_write: typing.Callable[[], T_Result], + ) -> T_Result: + if self is OauthScope.AUTHORIZED_USER_READ: + return authorized_user_read() + if self is OauthScope.READ_PAGES: + return read_pages() + if self is OauthScope.SITES_READ: + return sites_read() + if self is OauthScope.SITES_WRITE: + return sites_write() + if self is OauthScope.CUSTOM_CODE_READ: + return custom_code_read() + if self is OauthScope.CUSTOM_CODE_WRITE: + return custom_code_write() + if self is OauthScope.CUSTOM_CODE_DELETE: + return custom_code_delete() + if self is OauthScope.USERS_READ: + return users_read() + if self is OauthScope.USERS_WRITE: + return users_write() + if self is OauthScope.ECOMMERCE_READ: + return ecommerce_read() + if self is OauthScope.ECOMMERCE_WRITE: + return ecommerce_write() diff --git a/tests/test_client.py b/tests/test_client.py index 60a58e6..de1cc95 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,6 +1,7 @@ import pytest +import pytest # Get started with writing tests with pytest at https://docs.pytest.org @pytest.mark.skip(reason="Unimplemented") def test_client() -> None: - assert True == True + assert True == True \ No newline at end of file