🌿 Fern Regeneration -- support OAuth 2.0 Authentication

.fernignore

@@ -2,3 +2,6 @@
 
 README.md
 assets/
+
+src/webflow/client.py
+src/webflow/oauth.py

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! 

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",

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)

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 = "<YOUR_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 = "<YOUR_CLIENT_ID>",
+        client_secret = "<YOUR_CLIENT_ID>",
+        code= "<YOUR_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)
+

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",

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()