From d49c202ad01ac098183f7fa91e7cbb5fa2cb42d6 Mon Sep 17 00:00:00 2001 From: Koosha Owji Date: Sat, 11 Jul 2026 15:02:44 -0500 Subject: [PATCH 1/2] fix: update Python docs to match SDK --- .../sdks/backend/python-sdk.mdx | 468 +++++++----------- 1 file changed, 190 insertions(+), 278 deletions(-) diff --git a/src/content/docs/developer-tools/sdks/backend/python-sdk.mdx b/src/content/docs/developer-tools/sdks/backend/python-sdk.mdx index 1c0ac5fc7..d8fe6fc0e 100644 --- a/src/content/docs/developer-tools/sdks/backend/python-sdk.mdx +++ b/src/content/docs/developer-tools/sdks/backend/python-sdk.mdx @@ -83,7 +83,7 @@ login_url = asyncio.get_event_loop().run_until_complete(oauth.login()) 1. **Update imports**: Change from `kinde_sdk.kinde_api_client` to the appropriate client type 2. **Update initialization**: Use the new client initialization pattern 3. **Update method calls**: Most methods are now async - use `await` or `asyncio.run_until_complete()` -4. **Update error handling**: Use new exception types from `kinde_sdk.exceptions` +4. **Update error handling**: Use the exception types from `kinde_sdk.core.exceptions` (authentication) and `kinde_sdk.management.exceptions` (Management API) 5. **Test thoroughly**: Verify all authentication flows work correctly ### Client selection recommendations @@ -1386,239 +1386,184 @@ You don't need to manually manage tokens or sessions - the SDK handles this auto ## Management API -The Kinde Python SDK provides a Management API client for interacting with Kinde's management endpoints. This allows you to programmatically manage users, organizations, and other resources. The Management API supports both sync and async patterns. +The Kinde Python SDK includes a Management API client for programmatically managing users, organizations, roles, and other resources. It is a **synchronous** client that is separate from the `OAuth`/`AsyncOAuth`/`SmartOAuth` user-authentication clients — you create it directly with your machine-to-machine (M2M) application credentials, and its calls are not awaited. ### Getting started -#### With OAuth client (Framework-based) +Create a `ManagementClient` with your M2M application's credentials. It authenticates automatically using the client credentials grant. ```python -from kinde_sdk.auth.oauth import OAuth -from flask import Flask -import asyncio - -app = Flask(__name__) -oauth = OAuth(framework="flask", app=app) +from kinde_sdk.management import ManagementClient -# Get the management client -management = oauth.get_management() +management = ManagementClient( + domain="your-domain.kinde.com", + client_id="your-m2m-client-id", + client_secret="your-m2m-client-secret", +) -# Use with asyncio in Flask -def list_users_sync(): - loop = asyncio.get_event_loop() - users = loop.run_until_complete(management.get_users()) - return users +# List users +users = management.users_api.get_users(page_size=10) ``` -#### With AsyncOAuth client (Native async) +Each API group is available as a `_api` attribute, for example `management.users_api`, `management.organizations_api`, and `management.roles_api`. -```python -from kinde_sdk.auth.async_oauth import AsyncOAuth - -oauth = AsyncOAuth() +> The Management API requires a separate M2M application in Kinde with the relevant scopes enabled. It is not tied to a signed-in user, so it does not use your `OAuth` client or a user session. -# Get the management client (native async) -management = await oauth.get_management() - -# All methods are async -users = await management.get_users() -``` +### Available endpoints -#### With SmartOAuth client (Context-aware) +**User management:** ```python -from kinde_sdk.auth.smart_oauth import SmartOAuth +from kinde_sdk.management.models.create_user_request import CreateUserRequest +from kinde_sdk.management.models.create_user_request_profile import CreateUserRequestProfile +from kinde_sdk.management.models.create_user_request_identities_inner import CreateUserRequestIdentitiesInner +from kinde_sdk.management.models.update_user_request import UpdateUserRequest + +# List users +users = management.users_api.get_users(page_size=10) + +# Get a specific user +user = management.users_api.get_user_data(id="user_123") + +# Create a new user +new_user = management.users_api.create_user( + create_user_request=CreateUserRequest( + profile=CreateUserRequestProfile(given_name="John", family_name="Doe"), + identities=[ + CreateUserRequestIdentitiesInner( + type="email", + details={"email": "user@example.com"}, + ) + ], + ) +) -oauth = SmartOAuth() +# Update a user +updated_user = management.users_api.update_user( + id="user_123", + update_user_request=UpdateUserRequest(given_name="Johnny"), +) -# Works in async context -async def async_get_users(): - management = await oauth.get_management() - return await management.get_users() - -# Works in sync context (if supported) -def sync_get_users(): - management = oauth.get_management() - return management.get_users() +# Delete a user +management.users_api.delete_user(id="user_123") ``` -### Available endpoints - -The Management API provides methods for common operations on resources. All examples use async patterns: - -**User management:** +### Organization management ```python -# List users (async) -users = await management.get_users() +from kinde_sdk.management.models.create_organization_request import CreateOrganizationRequest +from kinde_sdk.management.models.update_organization_request import UpdateOrganizationRequest -# Get a specific user (async) -user = await management.get_user(user_id="user_123") +# List organizations +orgs = management.organizations_api.get_organizations(page_size=10) -# Create a new user (async) -new_user = await management.create_user( - email="user@example.com", - given_name="John", - family_name="Doe" +# Get a specific organization +org = management.organizations_api.get_organization(code="org_123") + +# Create a new organization +new_org = management.organizations_api.create_organization( + create_organization_request=CreateOrganizationRequest(name="My Organization"), ) -# Update a user (async) -updated_user = await management.update_user( - user_id="user_123", - given_name="Johnny" +# Update an organization +updated_org = management.organizations_api.update_organization( + org_code="org_123", + update_organization_request=UpdateOrganizationRequest(name="Updated Name"), ) -# Delete a user (async) -await management.delete_user(user_id="user_123") +# Delete an organization +management.organizations_api.delete_organization(org_code="org_123") ``` -### Organization management - -**Using Management API with FastAPI (OAuth client):** +**Organization invites:** ```python -from fastapi import FastAPI, HTTPException -from kinde_sdk.auth.oauth import OAuth - -app = FastAPI() -oauth = OAuth(framework="fastapi", app=app) - -@app.get("/organizations") -async def list_organizations(): - management = oauth.get_management() - orgs = await management.get_organizations() - return orgs - -@app.get("/organizations/{org_id}") -async def get_organization(org_id: str): - management = oauth.get_management() - org = await management.get_organization(org_id=org_id) - return org - -@app.post("/organizations") -async def create_organization(name: str): - management = oauth.get_management() - new_org = await management.create_organization(name=name) - return new_org - -@app.put("/organizations/{org_id}") -async def update_organization(org_id: str, name: str): - management = oauth.get_management() - updated_org = await management.update_organization( - org_id=org_id, - name=name - ) - return updated_org +from kinde_sdk.management.models.create_organization_invite_request import CreateOrganizationInviteRequest + +# List invites for an organization +invites = management.organizations_api.get_organization_invites(org_code="org_123") + +# Create an invite +new_invite = management.organizations_api.create_organization_invite( + org_code="org_123", + create_organization_invite_request=CreateOrganizationInviteRequest( + email="invitee@example.com", + first_name="Jane", + last_name="Doe", + roles=["member"], # role keys to assign on acceptance + ), +) -@app.delete("/organizations/{org_id}") -async def delete_organization(org_id: str): - management = oauth.get_management() - await management.delete_organization(org_id=org_id) - return {"message": "Organization deleted"} +# Delete an invite +management.organizations_api.delete_organization_invite( + org_code="org_123", invite_code="invite_123" +) ``` -**Using Management API with AsyncOAuth client:** +**Using the Management API in a FastAPI route:** ```python -from kinde_sdk.auth.async_oauth import AsyncOAuth +from fastapi import FastAPI +from kinde_sdk.management import ManagementClient -oauth = AsyncOAuth() +app = FastAPI() +management = ManagementClient( + domain="your-domain.kinde.com", + client_id="your-m2m-client-id", + client_secret="your-m2m-client-secret", +) -async def manage_organizations(): - # Get management client - management = await oauth.get_management() - - # List organizations - orgs = await management.get_organizations() - - # Get a specific organization - org = await management.get_organization(org_id="org_123") - - # Create a new organization - new_org = await management.create_organization( - name="My Organization" - ) - - # Update an organization - updated_org = await management.update_organization( - org_id="org_123", - name="Updated Name" - ) - - # Delete an organization - await management.delete_organization(org_id="org_123") - - return orgs +@app.get("/organizations") +async def list_organizations(): + # The management call itself is synchronous + return management.organizations_api.get_organizations() ``` ### Error handling -The Management API methods will raise exceptions for API errors. It's recommended to handle these appropriately: - -**Example with OAuth client (FastAPI):** +Management API methods raise `ApiException` (and typed subclasses such as `NotFoundException` and `UnauthorizedException`) from `kinde_sdk.management.exceptions`. The exception exposes `.status`, `.reason`, and `.body`. ```python from fastapi import FastAPI, HTTPException -from kinde_sdk.auth.oauth import OAuth -from kinde_sdk.exceptions import KindeAPIException +from kinde_sdk.management import ManagementClient +from kinde_sdk.management.exceptions import ApiException, NotFoundException app = FastAPI() -oauth = OAuth(framework="fastapi", app=app) +management = ManagementClient( + domain="your-domain.kinde.com", + client_id="your-m2m-client-id", + client_secret="your-m2m-client-secret", +) @app.get("/users/{user_id}") async def get_user(user_id: str): - management = oauth.get_management() - try: - user = await management.get_user(user_id=user_id) - return user - except KindeAPIException as e: - raise HTTPException(status_code=e.status_code, detail=str(e)) - except Exception as e: - raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}") -``` - -**Example with AsyncOAuth client:** - -```python -from kinde_sdk.auth.async_oauth import AsyncOAuth -from kinde_sdk.exceptions import KindeAPIException - -oauth = AsyncOAuth() - -async def get_user_safely(user_id: str): - management = await oauth.get_management() try: - user = await management.get_user(user_id=user_id) - return user - except KindeAPIException as e: - print(f"API Error {e.status_code}: {e.message}") - return None - except Exception as e: - print(f"Unexpected error: {str(e)}") - return None + return management.users_api.get_user_data(id=user_id) + except NotFoundException: + raise HTTPException(status_code=404, detail="User not found") + except ApiException as e: + raise HTTPException(status_code=e.status, detail=e.reason) ``` ### Token management -The Management API client has its own token management system for API authentication, which is separate from the core SDK's user session token management. The Management API client automatically handles: +The Management API client has its own token management, separate from the core SDK's user session token management. It automatically: -- **accessing Kinde Management API endpoints**: Obtains tokens for accessing Kinde's management endpoints -- **Token refresh**: Automatically refreshes management API tokens when they expire -- **Token storage**: Securely stores management API tokens -- **Thread safety**: Ensures thread-safe token handling for concurrent requests +- Obtains access tokens for Kinde's management endpoints using your M2M credentials +- Refreshes management tokens when they expire +- Caches tokens and shares them across clients with the same domain and client ID +- Handles concurrent requests safely -You don't need to manually manage Management API tokens - the client handles this for you. This is different from the core SDK's user session token management, which handles user authentication tokens automatically. +You don't need to manage these tokens yourself. ### Best practices -1. **Always use async/await when calling Management API methods**: The Management API is async-native for better performance -2. **Handle API errors appropriately**: Use try/except blocks and handle `KindeAPIException` specifically -3. **Cache results when appropriate**: Reduce API calls by caching user data, organizations, and permissions -4. **Use appropriate error handling for production**: Implement logging, monitoring, and graceful error recovery -5. **Keep your client credentials secure**: Use environment variables, never commit secrets to version control -6. **Use connection pooling**: For high-traffic applications, configure HTTP connection pooling -7. **Implement retry logic**: Add retry logic with exponential backoff for transient failures -8. **Monitor token expiration**: Handle token refresh gracefully to avoid authentication failures +1. **Use a dedicated M2M application**: Create a separate machine-to-machine application for the Management API with only the scopes it needs. +2. **Handle API errors appropriately**: Wrap calls in try/except and handle `ApiException` (and its subclasses) from `kinde_sdk.management.exceptions`. +3. **Cache results when appropriate**: Reduce API calls by caching data such as users, organizations, and roles. +4. **Keep your client credentials secure**: Use environment variables and never commit secrets to version control. +5. **Add retry logic for transient failures**: Retry with exponential backoff on network errors and 5xx responses. +6. **Reuse the client**: Create the `ManagementClient` once and reuse it so cached tokens are shared. For more information about the Management API endpoints and capabilities, see the [Kinde Management API documentation](https://docs.kinde.com/kinde-apis/management/). @@ -1628,109 +1573,96 @@ Proper error handling is crucial for building robust applications with the Kinde ### Common exceptions -The SDK raises specific exception types that you should handle: +The SDK raises specific exception types that you should handle. Authentication and +core errors come from `kinde_sdk.core.exceptions`; Management API errors come from +`kinde_sdk.management.exceptions`: ```python -from kinde_sdk.exceptions import ( - KindeAPIException, - KindeAuthenticationException, - KindeAuthorizationException, - KindeValidationException, - KindeConfigurationException +# Core / authentication exceptions +from kinde_sdk.core.exceptions import ( + KindeException, # base class for all core errors + KindeConfigurationException, # missing/invalid configuration + KindeLoginException, # login / callback failures + KindeTokenException, # token exchange / validation failures + KindeRetrieveException, # failure retrieving user or token data +) + +# Management API exceptions +from kinde_sdk.management.exceptions import ( + ApiException, # base class; exposes .status, .reason, .body + BadRequestException, # 400 + UnauthorizedException, # 401 + ForbiddenException, # 403 + NotFoundException, # 404 ) ``` ### Error handling patterns -**Pattern 1: Comprehensive error handling (FastAPI)** +**Pattern 1: Authentication check (FastAPI)** + +Authentication state is read synchronously and does not raise — `is_authenticated()` +returns a bool and `get_user_info()` returns the user's claims. Guard the route yourself: ```python from fastapi import FastAPI, HTTPException from kinde_sdk.auth.oauth import OAuth -from kinde_sdk.exceptions import ( - KindeAPIException, - KindeAuthenticationException, - KindeAuthorizationException -) app = FastAPI() oauth = OAuth(framework="fastapi", app=app) @app.get("/protected") -async def protected_route(request: Request): - try: - # Check authentication - if not await oauth.is_authenticated(request): - raise HTTPException(status_code=401, detail="Not authenticated") - - # Get user info - user_info = await oauth.get_user_info(request) - return user_info - - except KindeAuthenticationException as e: - # Handle authentication errors - raise HTTPException(status_code=401, detail=f"Authentication failed: {str(e)}") - - except KindeAuthorizationException as e: - # Handle authorization errors - raise HTTPException(status_code=403, detail=f"Authorization failed: {str(e)}") - - except KindeAPIException as e: - # Handle API errors - raise HTTPException(status_code=e.status_code, detail=f"API error: {str(e)}") - - except Exception as e: - # Handle unexpected errors - raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}") +async def protected_route(): + if not oauth.is_authenticated(): + raise HTTPException(status_code=401, detail="Not authenticated") + return oauth.get_user_info() ``` -**Pattern 2: Error handling with AsyncOAuth** +**Pattern 2: Handling Management API errors** + +Management API calls raise `ApiException` (and typed subclasses) from +`kinde_sdk.management.exceptions`. The exception exposes `.status` and `.reason`: ```python -from kinde_sdk.auth.async_oauth import AsyncOAuth -from kinde_sdk.exceptions import KindeAPIException import logging +from kinde_sdk.management import ManagementClient +from kinde_sdk.management.exceptions import ApiException, NotFoundException -oauth = AsyncOAuth() logger = logging.getLogger(__name__) +management = ManagementClient( + domain="your-domain.kinde.com", + client_id="your-m2m-client-id", + client_secret="your-m2m-client-secret", +) -async def safe_get_user(): +def safe_get_user(user_id: str): try: - user_info = await oauth.get_user_info() - return user_info - except KindeAPIException as e: - logger.error(f"Kinde API error: {e.status_code} - {e.message}") + return management.users_api.get_user_data(id=user_id) + except NotFoundException: return None - except Exception as e: - logger.exception(f"Unexpected error: {str(e)}") + except ApiException as e: + logger.error(f"Kinde Management API error {e.status}: {e.reason}") return None ``` -**Pattern 3: Permission checking with error handling** +**Pattern 3: Permission checking** + +Permission checks live in `kinde_sdk.auth.permissions` and are async. They return a +result dict rather than raising, so check `isGranted`: ```python from fastapi import HTTPException -from kinde_sdk.auth.oauth import OAuth -from kinde_sdk.exceptions import KindeAPIException - -oauth = OAuth(framework="fastapi", app=app) +from kinde_sdk.auth import permissions -async def check_permission_with_error_handling(request: Request, permission: str): - try: - perm_result = await oauth.get_permission(permission, request) - if not perm_result.get("isGranted"): - raise HTTPException( - status_code=403, - detail=f"Permission '{permission}' not granted" - ) - return perm_result - except KindeAPIException as e: +async def require_permission(permission_key: str): + result = await permissions.get_permission(permission_key) + if not result.get("isGranted"): raise HTTPException( - status_code=e.status_code, - detail=f"Error checking permission: {str(e)}" + status_code=403, + detail=f"Permission '{permission_key}' not granted", ) + return result ``` - ### Best practices summary #### Authentication and authorization @@ -1783,52 +1715,32 @@ async def check_permission_with_error_handling(request: Request, permission: str from fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse from kinde_sdk.auth.oauth import OAuth -from kinde_sdk.exceptions import ( - KindeAPIException, - KindeAuthenticationException, - KindeAuthorizationException, - KindeValidationException -) +from kinde_sdk.auth import permissions +from kinde_sdk.management.exceptions import ApiException import logging app = FastAPI() oauth = OAuth(framework="fastapi", app=app) logger = logging.getLogger(__name__) -# Global exception handler -@app.exception_handler(KindeAuthenticationException) -async def auth_exception_handler(request: Request, exc: KindeAuthenticationException): - logger.warning(f"Authentication failed: {str(exc)}") +# Convert Management API errors into JSON responses +@app.exception_handler(ApiException) +async def management_exception_handler(request: Request, exc: ApiException): + logger.error(f"Kinde Management API error {exc.status}: {exc.reason}") return JSONResponse( - status_code=401, - content={"error": "Authentication required", "detail": "Please log in"} + status_code=exc.status or 500, + content={"error": "Management API error", "detail": exc.reason}, ) -@app.exception_handler(KindeAuthorizationException) -async def authz_exception_handler(request: Request, exc: KindeAuthorizationException): - logger.warning(f"Authorization failed: {str(exc)}") - return JSONResponse( - status_code=403, - content={"error": "Access denied", "detail": "Insufficient permissions"} - ) +@app.get("/api/protected") +async def protected_endpoint(): + # Authentication is read synchronously and does not raise + if not oauth.is_authenticated(): + raise HTTPException(status_code=401, detail="Not authenticated") -@app.exception_handler(KindeAPIException) -async def api_exception_handler(request: Request, exc: KindeAPIException): - logger.error(f"Kinde API error {exc.status_code}: {str(exc)}") - return JSONResponse( - status_code=exc.status_code, - content={"error": "API error", "detail": "An error occurred with the authentication service"} - ) + result = await permissions.get_permission("read:data") + if not result.get("isGranted"): + raise HTTPException(status_code=403, detail="Permission denied") -@app.get("/api/protected") -async def protected_endpoint(request: Request): - # This will automatically handle exceptions via the handlers above - if not await oauth.is_authenticated(request): - raise KindeAuthenticationException("Not authenticated") - - permission = await oauth.get_permission("read:data", request) - if not permission.get("isGranted"): - raise KindeAuthorizationException("Permission denied") - return {"message": "Access granted"} ``` From 1c382bca638c6da9fa0bfd13170ff97e8bf04ab6 Mon Sep 17 00:00:00 2001 From: Koosha Owji Date: Thu, 3 Sep 2026 16:27:12 +1000 Subject: [PATCH 2/2] fix: address Python SDK doc feedback --- .../sdks/backend/python-sdk.mdx | 524 +++++++----------- 1 file changed, 211 insertions(+), 313 deletions(-) diff --git a/src/content/docs/developer-tools/sdks/backend/python-sdk.mdx b/src/content/docs/developer-tools/sdks/backend/python-sdk.mdx index d8fe6fc0e..d7f5fcb8e 100644 --- a/src/content/docs/developer-tools/sdks/backend/python-sdk.mdx +++ b/src/content/docs/developer-tools/sdks/backend/python-sdk.mdx @@ -30,7 +30,7 @@ keywords: - environment variables - callback URLs - session management -updated: 2026-01-23 +updated: 2026-09-03 featured: false deprecated: false ai_summary: Complete guide for Python SDK including Flask and FastAPI integration, OAuth configuration, environment variables, and session management for Python 3.9+ applications. @@ -322,8 +322,8 @@ async def lambda_handler(event, context): # Check authentication if event.get('path') == '/user': - if await oauth.is_authenticated(event): - user_info = await oauth.get_user_info(event) + if oauth.is_authenticated(): + user_info = await oauth.get_user_info_async() return { 'statusCode': 200, 'body': json.dumps(user_info) @@ -507,7 +507,7 @@ def logout(): def get_user(): """Get the current user's information.""" try: - if not oauth.is_authenticated(request): + if not oauth.is_authenticated(): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: @@ -515,7 +515,7 @@ def get_user(): return redirect(login_url) finally: loop.close() - return oauth.get_user_info(request) + return oauth.get_user_info() except Exception as e: return f"Failed to get user info: {str(e)}", 400 ``` @@ -568,10 +568,10 @@ async def logout(request: Request): @app.get("/user") async def get_user(request: Request): """Get the current user's information.""" - if not await oauth.is_authenticated(request): + if not oauth.is_authenticated(): url = await oauth.login() return RedirectResponse(url=url) - return await oauth.get_user_info(request) + return oauth.get_user_info() ``` **Using AsyncOAuth client with FastAPI (for more control):** @@ -666,7 +666,7 @@ def logout(): def get_user(): """Get the current user's information."" try: - if not oauth.is_authenticated(request): + if not oauth.is_authenticated(): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: @@ -675,7 +675,7 @@ def get_user(): finally: loop.close() - return oauth.get_user_info(request) + return oauth.get_user_info() except Exception as e: return f"Failed to get user info: {str(e)}", 400 ``` @@ -711,9 +711,9 @@ async def logout(request: Request): @app.get("/user") async def get_user(request: Request): - if not oauth.is_authenticated(request): + if not oauth.is_authenticated(): return RedirectResponse(url=await oauth.login()) - return oauth.get_user_info(request) + return oauth.get_user_info() ``` The manual implementation gives you more control over the authentication flow and allows you to add custom logic like session management, error handling, and logging. Note that Flask requires special handling of async methods using `asyncio` since it doesn't natively support async/await like FastAPI does. @@ -730,137 +730,66 @@ The manual implementation gives you more control over the authentication flow an ## User permissions -The Kinde Python SDK provides a simple way to check user permissions in your application. The API supports both sync and async patterns depending on your client type. - -### With OAuth client (Framework-based) - -```python -from kinde_sdk.auth.oauth import OAuth -from flask import request -import asyncio - -oauth = OAuth(framework="flask", app=app) - -# Async pattern (required for OAuth client) -def check_permission_sync(): - loop = asyncio.get_event_loop() - permission = loop.run_until_complete( - oauth.get_permission("create:todos", request) - ) - return permission["isGranted"] - -# In FastAPI (native async) -@app.get("/todos") -async def create_todo(request: Request): - permission = await oauth.get_permission("create:todos", request) - if not permission["isGranted"]: - raise HTTPException(status_code=403, detail="Permission denied") - # Create todo logic... -``` - -### With AsyncOAuth client (Native async) +The Kinde Python SDK provides a simple way to check user permissions in your application. Permission checks use the `permissions` helper from `kinde_sdk.auth`. It reads the signed-in user's session directly, so it works the same way whichever OAuth client (`OAuth`, `AsyncOAuth`, or `SmartOAuth`) you use. You still create an OAuth client for your framework as usual; the helper reads the user session that client manages. Its methods are async. -```python -from kinde_sdk.auth.async_oauth import AsyncOAuth - -oauth = AsyncOAuth() - -# Native async pattern -async def check_permission(): - permission = await oauth.get_permission("create:todos") - if permission["isGranted"]: - print(f"User has permission in organization: {permission['orgCode']}") - return True - return False - -# Get all permissions -async def get_all_permissions(): - all_permissions = await oauth.get_permissions() - print(f"User belongs to organization: {all_permissions['orgCode']}") - print("User permissions:", all_permissions["permissions"]) - return all_permissions -``` - -### With SmartOAuth client (Context-aware) +### Checking permissions ```python -from kinde_sdk.auth.smart_oauth import SmartOAuth +from kinde_sdk.auth import permissions -oauth = SmartOAuth() +# Check a single permission +permission = await permissions.get_permission("create:todos") +if permission["isGranted"]: + print(f"User can create todos in organization: {permission['orgCode']}") -# Works in async context -async def async_check(): - permission = await oauth.get_permission("create:todos") - return permission["isGranted"] - -# Works in sync context (if available) -def sync_check(): - permission = oauth.get_permission("create:todos") - return permission["isGranted"] +# Get all permissions for the current user +all_permissions = await permissions.get_permissions() +print(f"User belongs to organization: {all_permissions['orgCode']}") +print("User permissions:", all_permissions["permissions"]) ``` -### Checking permissions +`get_permission` returns a dict with `permissionKey`, `orgCode`, and `isGranted`. `get_permissions` returns a dict with `orgCode` and a `permissions` list. If the user is not authenticated, the helpers do not raise: `isGranted` is `False` and `permissions` is an empty list. ### Practical examples -Here's how to use permissions in your application with different client types: +**Example 1: Permission check in FastAPI** -**Example 1: Permission check in FastAPI (OAuth client)** +FastAPI routes are async, so you can await the helper directly: ```python -from fastapi import FastAPI, Request, HTTPException +from fastapi import FastAPI, HTTPException from kinde_sdk.auth.oauth import OAuth +from kinde_sdk.auth import permissions app = FastAPI() oauth = OAuth(framework="fastapi", app=app) @app.post("/todos") -async def create_todo(request: Request, todo_data: dict): - permission = await oauth.get_permission("create:todos", request) +async def create_todo(todo_data: dict): + permission = await permissions.get_permission("create:todos") if not permission["isGranted"]: raise HTTPException(status_code=403, detail="Permission denied") # Create todo logic here... return {"message": "Todo created"} @app.get("/todos") -async def list_todos(request: Request): - permission = await oauth.get_permission("read:todos", request) +async def list_todos(): + permission = await permissions.get_permission("read:todos") if not permission["isGranted"]: raise HTTPException(status_code=403, detail="Permission denied") # List todos logic... return {"todos": []} ``` -**Example 2: Permission check with AsyncOAuth client** - -```python -from kinde_sdk.auth.async_oauth import AsyncOAuth - -oauth = AsyncOAuth() - -async def create_todo_handler(request): - permission = await oauth.get_permission("create:todos") - if not permission["isGranted"]: - return {"error": "Permission denied"}, 403 - - org_code = permission.get("orgCode") - # Create todo with organization context - return {"message": "Todo created", "org_code": org_code} - -async def get_all_user_permissions(): - all_permissions = await oauth.get_permissions() - return { - "org_code": all_permissions["orgCode"], - "permissions": all_permissions["permissions"] - } -``` +**Example 2: Permission check in Flask** -**Example 3: Permission-based conditional rendering (Flask)** +Flask routes are synchronous, so run the async helper with `asyncio`: ```python -from flask import Flask, request, render_template -from kinde_sdk.auth.oauth import OAuth import asyncio +from flask import Flask, render_template +from kinde_sdk.auth.oauth import OAuth +from kinde_sdk.auth import permissions app = Flask(__name__) oauth = OAuth(framework="flask", app=app) @@ -869,158 +798,157 @@ oauth = OAuth(framework="flask", app=app) def dashboard(): loop = asyncio.get_event_loop() can_create = loop.run_until_complete( - oauth.get_permission("create:todos", request) + permissions.get_permission("create:todos") )["isGranted"] - + can_delete = loop.run_until_complete( - oauth.get_permission("delete:todos", request) + permissions.get_permission("delete:todos") )["isGranted"] - - return render_template("dashboard.html", - can_create=can_create, + + return render_template("dashboard.html", + can_create=can_create, can_delete=can_delete) ``` +**Example 3: Organization context** + +```python +from kinde_sdk.auth import permissions + +async def create_todo_handler(): + permission = await permissions.get_permission("create:todos") + if not permission["isGranted"]: + return {"error": "Permission denied"}, 403 + + org_code = permission.get("orgCode") + # Create todo with organization context + return {"message": "Todo created", "org_code": org_code} + +async def get_all_user_permissions(): + all_permissions = await permissions.get_permissions() + return { + "org_code": all_permissions["orgCode"], + "permissions": all_permissions["permissions"] + } +``` + ### Common permission patterns Here are some common permission patterns you might use: ```python # Resource-based permissions -"create:todos -"read:todos -"update:todos -"delete:todos +"create:todos" +"read:todos" +"update:todos" +"delete:todos" # Feature-based permissions -"can:export_data -"can:manage_users -"can:view_analytics +"can:export_data" +"can:manage_users" +"can:view_analytics" # Organization-based permissions -"org:manage_members -"org:view_billing -"org:update_settings +"org:manage_members" +"org:view_billing" +"org:update_settings" ``` For more information about setting up permissions in Kinde, see [User permissions](/manage-users/roles-and-permissions/user-permissions/). ## Feature flags -The Kinde Python SDK provides a simple way to access feature flags from your application. Feature flags support both sync and async patterns. +The Kinde Python SDK provides a simple way to access feature flags from your application. Feature flags are read with the `feature_flags` helper from `kinde_sdk.auth`. Like the permissions helper, it reads the signed-in user's session directly, works with any OAuth client, and its methods are async. -### With OAuth client (Framework-based) +### Getting feature flags ```python -from kinde_sdk.auth.oauth import OAuth -from flask import request -import asyncio +from kinde_sdk.auth import feature_flags -oauth = OAuth(framework="flask", app=app) +# Get a string feature flag +theme_flag = await feature_flags.get_flag("theme", default_value="light") +print(f"Current theme: {theme_flag.value}") -# Async pattern (required for OAuth client) -def get_theme_sync(): - loop = asyncio.get_event_loop() - theme_flag = loop.run_until_complete( - oauth.get_flag("theme", request, default_value="light") - ) - return theme_flag.value - -# In FastAPI (native async) -@app.get("/settings") -async def get_settings(request: Request): - theme = await oauth.get_flag("theme", request, default_value="light") - dark_mode = await oauth.get_flag("is_dark_mode", request, default_value=False) - return { - "theme": theme.value, - "dark_mode": dark_mode.value - } -``` - -### With AsyncOAuth client (Native async) - -```python -from kinde_sdk.auth.async_oauth import AsyncOAuth +# Get a boolean feature flag with default value +dark_mode = await feature_flags.get_flag("is_dark_mode", default_value=False) +if dark_mode.value: + print("Dark mode is enabled") -oauth = AsyncOAuth() +# Get a numeric feature flag +competitions_limit = await feature_flags.get_flag("competitions_limit", default_value=3) +print(f"User can create up to {competitions_limit.value} competitions") -# Native async pattern -async def get_feature_flags(): - # Get a string feature flag - theme_flag = await oauth.get_flag("theme", default_value="light") - print(f"Current theme: {theme_flag.value}") - - # Get a boolean feature flag with default value - dark_mode = await oauth.get_flag("is_dark_mode", default_value=False) - if dark_mode.value: - print("Dark mode is enabled") - - # Get a numeric feature flag - competitions_limit = await oauth.get_flag("competitions_limit", default_value=3) - print(f"User can create up to {competitions_limit.value} competitions") - - return { - "theme": theme_flag.value, - "dark_mode": dark_mode.value, - "limit": competitions_limit.value - } +# Get all feature flags +all_flags = await feature_flags.get_all_flags() +for code, flag in all_flags.items(): + print(f"{code}: {flag.value} ({flag.type})") ``` -### Getting feature flags - -To get a specific feature flag value: +`get_flag` returns a `FeatureFlag` object with `code`, `type` (`"string"`, `"boolean"`, or `"integer"`), `value`, and `is_default`. If the flag is not set for the user, or the user is not authenticated, `value` is the `default_value` you passed and `is_default` is `True`. `get_all_flags` returns a dict mapping flag codes to `FeatureFlag` objects. ### Practical examples -Here's how to use feature flags in your application with different client types: - -**Example 1: Conditional feature rendering (FastAPI with OAuth)** +**Example 1: Conditional feature rendering (FastAPI)** ```python -from fastapi import FastAPI, Request +from fastapi import FastAPI from kinde_sdk.auth.oauth import OAuth +from kinde_sdk.auth import feature_flags app = FastAPI() oauth = OAuth(framework="fastapi", app=app) @app.get("/competitions/create-button") -async def render_create_button(request: Request): - can_create = await oauth.get_flag("create_competition", request, default_value=False) +async def render_create_button(): + can_create = await feature_flags.get_flag("create_competition", default_value=False) if can_create.value: return {"html": ""} return {"html": ""} ``` -**Example 2: Theme configuration (AsyncOAuth)** +**Example 2: Theme configuration (Flask)** + +Flask routes are synchronous, so run the async helper with `asyncio`: ```python -from kinde_sdk.auth.async_oauth import AsyncOAuth +import asyncio +from flask import Flask +from kinde_sdk.auth.oauth import OAuth +from kinde_sdk.auth import feature_flags -oauth = AsyncOAuth() +app = Flask(__name__) +oauth = OAuth(framework="flask", app=app) -async def get_user_theme(): - theme = await oauth.get_flag("theme", default_value="light") - dark_mode = await oauth.get_flag("is_dark_mode", default_value=False) +@app.route("/settings") +def get_settings(): + loop = asyncio.get_event_loop() + theme = loop.run_until_complete( + feature_flags.get_flag("theme", default_value="light") + ) + dark_mode = loop.run_until_complete( + feature_flags.get_flag("is_dark_mode", default_value=False) + ) return { "theme": theme.value, - "is_dark_mode": dark_mode.value + "dark_mode": dark_mode.value } ``` **Example 3: Feature limits with validation (FastAPI)** ```python -from fastapi import FastAPI, Request, HTTPException +from fastapi import FastAPI, HTTPException from kinde_sdk.auth.oauth import OAuth +from kinde_sdk.auth import feature_flags app = FastAPI() oauth = OAuth(framework="fastapi", app=app) @app.post("/competitions") -async def create_competition(request: Request, competition_data: dict): - limit_flag = await oauth.get_flag("competitions_limit", request, default_value=3) - current_count = await get_user_competition_count(request) - +async def create_competition(competition_data: dict): + limit_flag = await feature_flags.get_flag("competitions_limit", default_value=3) + current_count = await get_user_competition_count() # Your implementation + if current_count >= limit_flag.value: raise HTTPException( status_code=403, @@ -1030,16 +958,13 @@ async def create_competition(request: Request, competition_data: dict): return {"message": "Competition created"} ``` -**Example 4: Type-safe flag access (AsyncOAuth)** +**Example 4: Inspecting all flags** ```python -from kinde_sdk.auth.async_oauth import AsyncOAuth - -oauth = AsyncOAuth() +from kinde_sdk.auth import feature_flags async def get_all_flags(): - # Get all feature flags - all_flags = await oauth.get_all_flags() + all_flags = await feature_flags.get_all_flags() result = {} for code, flag in all_flags.items(): result[code] = { @@ -1048,16 +973,6 @@ async def get_all_flags(): "is_default": flag.is_default } return result - -# Type-specific getters -async def get_boolean_flag(flag_name: str, default: bool = False): - return await oauth.get_boolean_flag(flag_name, default_value=default) - -async def get_string_flag(flag_name: str, default: str = ""): - return await oauth.get_string_flag(flag_name, default_value=default) - -async def get_integer_flag(flag_name: str, default: int = 0): - return await oauth.get_integer_flag(flag_name, default_value=default) ``` ### Feature flag types @@ -1103,83 +1018,46 @@ test_group = await feature_flags.get_flag("ab_test_group", default_value="contro ## Claims -The Kinde Python SDK provides a simple way to access user claims from your application. Claims support both sync and async patterns. +The Kinde Python SDK provides a simple way to access user claims from your application. Claims are read with the `claims` helper from `kinde_sdk.auth`. Like the other helpers, it reads the signed-in user's session directly, works with any OAuth client, and its methods are async. -### With OAuth client (Framework-based) +### Getting claims ```python -from kinde_sdk.auth.oauth import OAuth -from flask import request -import asyncio +from kinde_sdk.auth import claims -oauth = OAuth(framework="flask", app=app) +# Get the audience claim from the access token +aud_claim = await claims.get_claim("aud") +print(f"Token audience: {aud_claim['value']}") -# Async pattern (required for OAuth client) -def get_user_name_sync(): - loop = asyncio.get_event_loop() - claim = loop.run_until_complete( - oauth.get_claim("given_name", request, token_type="id_token") - ) - return claim["value"] +# Get the given_name claim from the ID token +name_claim = await claims.get_claim("given_name", token_type="id_token") +print(f"User's given name: {name_claim['value']}") -# In FastAPI (native async) -@app.get("/profile") -async def get_profile(request: Request): - given_name = await oauth.get_claim("given_name", request, token_type="id_token") - family_name = await oauth.get_claim("family_name", request, token_type="id_token") - email = await oauth.get_claim("email", request, token_type="id_token") - return { - "name": f"{given_name['value']} {family_name['value']}", - "email": email["value"] - } +# Get all claims from a token +all_claims = await claims.get_all_claims() +id_token_claims = await claims.get_all_claims(token_type="id_token") ``` -### With AsyncOAuth client (Native async) - -```python -from kinde_sdk.auth.async_oauth import AsyncOAuth - -oauth = AsyncOAuth() - -# Native async pattern -async def get_claims(): - # Get the audience claim from the access token - aud_claim = await oauth.get_claim("aud") - print(f"Token audience: {aud_claim['value']}") - - # Get the given_name claim from the ID token - name_claim = await oauth.get_claim("given_name", token_type="id_token") - print(f"User's given name: {name_claim['value']}") - - return { - "audience": aud_claim["value"], - "name": name_claim["value"] - } -``` - -### Getting claims - -To get a specific claim from the user's tokens: +`get_claim` returns a dict with `name` and `value`. If the claim is missing or the user is not authenticated, `value` is `None`. `get_all_claims` returns the token's claims as a dict. Both accept `token_type="access_token"` (the default) or `token_type="id_token"`. ### Practical examples -Here's how to use claims in your application with different client types: - -**Example 1: Accessing user information (FastAPI with OAuth)** +**Example 1: Accessing user information (FastAPI)** ```python -from fastapi import FastAPI, Request +from fastapi import FastAPI from kinde_sdk.auth.oauth import OAuth +from kinde_sdk.auth import claims app = FastAPI() oauth = OAuth(framework="fastapi", app=app) @app.get("/profile") -async def get_user_profile(request: Request): - given_name = await oauth.get_claim("given_name", request, token_type="id_token") - family_name = await oauth.get_claim("family_name", request, token_type="id_token") - email = await oauth.get_claim("email", request, token_type="id_token") - +async def get_user_profile(): + given_name = await claims.get_claim("given_name", token_type="id_token") + family_name = await claims.get_claim("family_name", token_type="id_token") + email = await claims.get_claim("email", token_type="id_token") + if given_name["value"] and family_name["value"]: return { "name": f"{given_name['value']} {family_name['value']}", @@ -1188,68 +1066,88 @@ async def get_user_profile(request: Request): return None ``` -**Example 2: Token validation (AsyncOAuth)** +**Example 2: Accessing user information (Flask)** + +Flask routes are synchronous, so run the async helper with `asyncio`: ```python -from fastapi import HTTPException -from kinde_sdk.auth.async_oauth import AsyncOAuth +import asyncio +from flask import Flask +from kinde_sdk.auth.oauth import OAuth +from kinde_sdk.auth import claims -oauth = AsyncOAuth() +app = Flask(__name__) +oauth = OAuth(framework="flask", app=app) -async def validate_token(): - aud_claim = await oauth.get_claim("aud") - if not aud_claim["value"] or "api.yourapp.com" not in aud_claim["value"]: - raise HTTPException(status_code=401, detail="Invalid token audience") - return {"message": "Access granted"} +@app.route("/profile") +def get_profile(): + loop = asyncio.get_event_loop() + given_name = loop.run_until_complete( + claims.get_claim("given_name", token_type="id_token") + ) + email = loop.run_until_complete( + claims.get_claim("email", token_type="id_token") + ) + return { + "name": given_name["value"], + "email": email["value"] + } +``` + +**Example 3: Token validation (FastAPI)** + +```python +from fastapi import FastAPI, HTTPException +from kinde_sdk.auth.oauth import OAuth +from kinde_sdk.auth import claims + +app = FastAPI() +oauth = OAuth(framework="fastapi", app=app) @app.get("/api/protected") async def protected_endpoint(): - return await validate_token() + aud_claim = await claims.get_claim("aud") + if not aud_claim["value"] or "api.yourapp.com" not in aud_claim["value"]: + raise HTTPException(status_code=401, detail="Invalid token audience") + return {"message": "Access granted"} ``` -**Example 3: Getting all claims (AsyncOAuth)** +**Example 4: Getting all claims** ```python -from kinde_sdk.auth.async_oauth import AsyncOAuth - -oauth = AsyncOAuth() +from kinde_sdk.auth import claims async def get_all_user_claims(): # Get all claims from the access token - all_claims = await oauth.get_all_claims() - access_token_data = {} - for claim_name, claim_value in all_claims.items(): - access_token_data[claim_name] = claim_value - + access_token_claims = await claims.get_all_claims() + # Get all claims from the ID token - id_token_claims = await oauth.get_all_claims(token_type="id_token") - id_token_data = {} - for claim_name, claim_value in id_token_claims.items(): - id_token_data[claim_name] = claim_value - + id_token_claims = await claims.get_all_claims(token_type="id_token") + return { - "access_token": access_token_data, - "id_token": id_token_data + "access_token": access_token_claims, + "id_token": id_token_claims } ``` -**Example 4: Organization context (FastAPI)** +**Example 5: Organization context (FastAPI)** ```python -from fastapi import FastAPI, Request +from fastapi import FastAPI from kinde_sdk.auth.oauth import OAuth +from kinde_sdk.auth import claims app = FastAPI() oauth = OAuth(framework="fastapi", app=app) @app.get("/organization-info") -async def get_org_info(request: Request): - org_code = await oauth.get_claim("org_code", request) - org_name = await oauth.get_claim("org_name", request) - +async def get_org_info(): + org_code = await claims.get_claim("org_code") + org_name = await claims.get_claim("org_name") + return { "org_code": org_code["value"], - "org_name": org_name["value"] if org_name else None + "org_name": org_name["value"] } ``` @@ -1259,10 +1157,10 @@ Here are some common claims you might want to access: ```python # User Information (ID Token) -"given_name -"family_name -"email -"picture +"given_name" +"family_name" +"email" +"picture" # Token Information (Access Token) "aud" # Audience @@ -1271,13 +1169,11 @@ Here are some common claims you might want to access: "iat" # Issued at time # Organization Information -"org_code -"org_name -"org_id +"org_code" +"org_name" +"org_id" ``` - - ## Organizations ### Create an organization @@ -1502,6 +1398,8 @@ management.organizations_api.delete_organization_invite( **Using the Management API in a FastAPI route:** +Declare routes that call the Management API with `def` rather than `async def`. The client is synchronous, and FastAPI runs sync routes in a threadpool, so the blocking HTTP call doesn't stall the event loop. If a route must be `async def`, offload the call with `await asyncio.to_thread(...)` instead. + ```python from fastapi import FastAPI from kinde_sdk.management import ManagementClient @@ -1514,8 +1412,8 @@ management = ManagementClient( ) @app.get("/organizations") -async def list_organizations(): - # The management call itself is synchronous +def list_organizations(): + # Sync route: FastAPI runs this in a threadpool return management.organizations_api.get_organizations() ``` @@ -1536,7 +1434,7 @@ management = ManagementClient( ) @app.get("/users/{user_id}") -async def get_user(user_id: str): +def get_user(user_id: str): try: return management.users_api.get_user_data(id=user_id) except NotFoundException: