From 02ad9f8d94e57586f37389881d9ebf8459ee33be Mon Sep 17 00:00:00 2001 From: brandtkruger Date: Mon, 12 Jan 2026 18:51:07 +0200 Subject: [PATCH] Enhance Python SDK documentation with migration guide and client selection recommendations Added a comprehensive migration guide for transitioning from v1 to v2 of the Kinde Python SDK, including code examples and client selection criteria. Expanded sections on using different client types (OAuth, AsyncOAuth, SmartOAuth) with practical implementation examples for Flask and FastAPI. Updated error handling and feature flag usage sections to reflect new patterns and best practices. --- .../sdks/backend/python-sdk.mdx | 1315 +++++++++++++++-- 1 file changed, 1180 insertions(+), 135 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 5b0eb652e..3a175dac0 100644 --- a/src/content/docs/developer-tools/sdks/backend/python-sdk.mdx +++ b/src/content/docs/developer-tools/sdks/backend/python-sdk.mdx @@ -45,10 +45,60 @@ The Kinde Python SDK allows developers to quickly and securely integrate a new o If you are using a previous version of Python, you may need to refer to the [previous v1 SDK](/developer-tools/sdks/backend/python-sdk-v1/). -If you're migrating from an older version of the SDK, see our [migration guide](https://github.com/kinde-oss/kinde-python-sdk/blob/main/MIGRATION.md) for detailed instructions. - For new projects, you can find our [Starter Kit on GitHub](https://github.com/kinde-starter-kits/python-starter-kit). +## Migration guide + +If you're migrating from an older version of the SDK or switching between client types, follow these recommendations: + +### Migrating from v1 SDK + +The current SDK (v2+) provides a more flexible architecture with multiple client types: + +**Old v1 SDK code:** +```python +from kinde_sdk.kinde_api_client import KindeApiClient + +kinde_client = KindeApiClient(**kinde_api_client_params) +login_url = kinde_client.get_login_url() +``` + +**New SDK code (OAuth client):** +```python +from kinde_sdk.auth.oauth import OAuth +import asyncio + +oauth = OAuth(framework="flask", app=app) +login_url = asyncio.get_event_loop().run_until_complete(oauth.login()) +``` + +### Choosing the right client for migration + +1. **If you're using Flask/FastAPI**: Continue using `OAuth` client +2. **If you're building async-only applications**: Migrate to `AsyncOAuth` +3. **If you need flexibility**: Consider `SmartOAuth` for context-aware behavior + +### Migration steps + +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` +5. **Test thoroughly**: Verify all authentication flows work correctly + +### Client selection recommendations + +| Scenario | Recommended Client | Reason | +|----------|-------------------|--------| +| Flask application | `OAuth` | Framework integration, session management | +| FastAPI application | `OAuth` or `AsyncOAuth` | Both work well, choose based on preference | +| Pure async app | `AsyncOAuth` | Native async support, better performance | +| Serverless/Lambda | `AsyncOAuth` | No framework dependencies, async-friendly | +| Library/Utility | `SmartOAuth` | Works in both sync and async contexts | +| Mixed codebase | `SmartOAuth` | Context-aware behavior | + +For more detailed migration instructions, see our [GitHub migration guide](https://github.com/kinde-oss/kinde-python-sdk/blob/main/MIGRATION.md). + ## Install Install [PIP](https://pip.pypa.io/en/stable/installation/) and then execute the following command: @@ -120,9 +170,46 @@ TEMPLATES_AUTO_RELOAD=True Kinde comes with a production environment, but you can set up other environments if you want to. Note that each environment needs to be set up independently, so you need to use the Environment subdomain in the code block above for those new environments. +## Choose your OAuth client + +The Kinde Python SDK provides three client types to suit different application needs: + +### Client selection guide + +**Use `OAuth` client when:** +- You're using Flask or FastAPI frameworks +- You want framework-specific integrations (automatic route registration, session management) +- Your application follows traditional request-response patterns +- You need synchronous operations with async support via framework adapters + +**Use `AsyncOAuth` client when:** +- You're building a native async application (e.g., pure async Python, async web frameworks) +- You want direct async/await support without framework abstractions +- You're using async libraries that don't fit traditional framework patterns +- You need maximum performance with async I/O operations + +**Use `SmartOAuth` client when:** +- Your application uses both sync and async contexts +- You want context-aware behavior (automatically sync or async based on usage) +- You're building libraries or utilities that need to work in various contexts +- You need flexibility to work with both sync and async code + +### Comparison table + +| Feature | OAuth | AsyncOAuth | SmartOAuth | +|---------|-------|------------|------------| +| Framework support | Flask, FastAPI | Any async framework | Any context | +| Sync operations | ✅ | ❌ | ✅ (context-aware) | +| Async operations | ✅ (via adapters) | ✅ (native) | ✅ (native) | +| Route registration | ✅ Automatic | ❌ Manual | ❌ Manual | +| Session management | ✅ Built-in | ⚠️ Custom | ⚠️ Custom | +| Best for | Framework apps | Pure async apps | Mixed contexts | + ## Configure your app -The OAuth client is now automatically configured based on the framework you're using. Simply import the OAuth class from the auth module and create an instance: +### Using the OAuth client (Framework-based) + +The `OAuth` client is automatically configured based on the framework you're using. Simply import the OAuth class from the auth module and create an instance: ```python from kinde_sdk.auth.oauth import OAuth @@ -146,7 +233,176 @@ oauth = OAuth( The SDK will automatically detect and configure the appropriate framework implementation based on the framework parameter and app instance you provide. -## Sign in and sign up +### Using the AsyncOAuth client (Native async) + +The `AsyncOAuth` client is designed for native async applications. It provides direct async/await support without framework abstractions: + +```python +from kinde_sdk.auth.async_oauth import AsyncOAuth + +# Initialize AsyncOAuth client +oauth = AsyncOAuth() + +# All methods are async +login_url = await oauth.login() +register_url = await oauth.register() +``` + +**Key features:** +- Native async/await support +- No framework dependencies +- Direct control over async operations +- Ideal for pure async applications and serverless functions + +### Using the SmartOAuth client (Context-aware) + +The `SmartOAuth` client automatically adapts to sync or async contexts: + +```python +from kinde_sdk.auth.smart_oauth import SmartOAuth + +# Initialize SmartOAuth client +oauth = SmartOAuth() + +# Works in async context +async def async_function(): + login_url = await oauth.login() # Returns coroutine + +# Works in sync context +def sync_function(): + login_url = oauth.login() # Returns URL directly +``` + +**Key features:** +- Context-aware behavior (sync/async) +- Works in both sync and async code +- Flexible usage patterns +- Ideal for libraries and utilities + +## Standalone usage (Serverless/Lambda) + +For serverless functions (AWS Lambda, Google Cloud Functions, Azure Functions), you can use the SDK in standalone mode without framework dependencies. + +### AWS Lambda example + +```python +import json +from kinde_sdk.auth.async_oauth import AsyncOAuth + +# Initialize client (can be outside handler for reuse) +oauth = AsyncOAuth() + +async def lambda_handler(event, context): + """AWS Lambda handler for authentication.""" + try: + # Handle login + if event.get('path') == '/login': + login_url = await oauth.login() + return { + 'statusCode': 302, + 'headers': {'Location': login_url} + } + + # Handle callback + if event.get('path') == '/callback': + code = event.get('queryStringParameters', {}).get('code') + state = event.get('queryStringParameters', {}).get('state') + + if not code: + return { + 'statusCode': 400, + 'body': json.dumps({'error': 'No code provided'}) + } + + result = await oauth.handle_redirect(code, state) + return { + 'statusCode': 200, + 'body': json.dumps({'message': 'Authenticated successfully'}) + } + + # Check authentication + if event.get('path') == '/user': + if await oauth.is_authenticated(event): + user_info = await oauth.get_user_info(event) + return { + 'statusCode': 200, + 'body': json.dumps(user_info) + } + else: + login_url = await oauth.login() + return { + 'statusCode': 302, + 'headers': {'Location': login_url} + } + + except Exception as e: + return { + 'statusCode': 500, + 'body': json.dumps({'error': str(e)}) + } +``` + +### Google Cloud Functions example + +```python +from flask import Request +from kinde_sdk.auth.async_oauth import AsyncOAuth +import asyncio + +oauth = AsyncOAuth() + +def cloud_function_handler(request: Request): + """Google Cloud Function handler.""" + async def handle_request(): + if request.path == '/login': + url = await oauth.login() + return redirect(url, code=302) + + if request.path == '/callback': + code = request.args.get('code') + state = request.args.get('state') + await oauth.handle_redirect(code, state) + return {'message': 'Authenticated'}, 200 + + return asyncio.run(handle_request()) +``` + +### Token storage for serverless + +In serverless environments, you'll need to implement custom token storage: + +```python +from kinde_sdk.auth.async_oauth import AsyncOAuth + +class ServerlessTokenStorage: + """Custom token storage for serverless environments.""" + + def __init__(self, storage_backend): # e.g., Redis, DynamoDB, etc. + self.storage = storage_backend + + async def store_tokens(self, user_id: str, tokens: dict): + await self.storage.set(f"tokens:{user_id}", tokens, ttl=3600) + + async def get_tokens(self, user_id: str) -> dict: + return await self.storage.get(f"tokens:{user_id}") + + async def clear_tokens(self, user_id: str): + await self.storage.delete(f"tokens:{user_id}") + +# Usage +storage = ServerlessTokenStorage(redis_client) +oauth = AsyncOAuth(token_storage=storage) +``` + +## Authentication flow + +The authentication flow consists of several steps. Here's a comprehensive guide to implementing authentication with the Kinde Python SDK. + +### Step 1: Initialize the client + +Choose the appropriate client based on your application type (see [Client Selection Guide](#choose-your-oauth-client)). + +### Step 2: Sign in and sign up The Kinde client provides methods for easy sign in and sign up. You can add buttons in your HTML as follows: @@ -185,6 +441,163 @@ oauth = OAuth( ) ``` +### Framework-specific implementation guides + +#### Flask implementation + +Flask applications use the `OAuth` client with framework support, or `AsyncOAuth` for async patterns. + +**Using OAuth client (recommended for Flask):** + +```python +from flask import Flask, request, session, redirect, url_for +from kinde_sdk.auth.oauth import OAuth +import asyncio + +app = Flask(__name__) +app.secret_key = 'your-secret-key' + +oauth = OAuth( + framework="flask", + app=app +) + +@app.route('/login') +def login(): + """Redirect to Kinde login page.""" + loop = asyncio.get_event_loop() + login_url = loop.run_until_complete(oauth.login()) + return redirect(login_url) + +@app.route('/register') +def register(): + """Redirect to Kinde registration page.""" + loop = asyncio.get_event_loop() + register_url = loop.run_until_complete(oauth.register()) + return redirect(register_url) + +@app.route('/callback') +def callback(): + """Handle the OAuth callback from Kinde.""" + try: + code = request.args.get('code') + state = request.args.get('state') + + if not code: + return "Authentication failed: No code provided", 400 + + user_id = session.get('user_id') or str(uuid.uuid4()) + loop = asyncio.get_event_loop() + result = loop.run_until_complete(oauth.handle_redirect(code, user_id, state)) + session['user_id'] = user_id + return redirect(url_for('index')) + except Exception as e: + return f"Authentication failed: {str(e)}", 400 + +@app.route('/logout') +def logout(): + """Logout the user and redirect to Kinde logout page.""" + user_id = session.get('user_id') + session.clear() + loop = asyncio.get_event_loop() + logout_url = loop.run_until_complete(oauth.logout(user_id)) + return redirect(logout_url) + +@app.route('/user') +def get_user(): + """Get the current user's information.""" + try: + if not oauth.is_authenticated(request): + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + login_url = loop.run_until_complete(oauth.login()) + return redirect(login_url) + finally: + loop.close() + return oauth.get_user_info(request) + except Exception as e: + return f"Failed to get user info: {str(e)}", 400 +``` + +#### FastAPI implementation + +FastAPI natively supports async/await, making it ideal for async operations. + +**Using OAuth client with FastAPI:** + +```python +from fastapi import FastAPI, Request, HTTPException +from fastapi.responses import RedirectResponse +from kinde_sdk.auth.oauth import OAuth + +app = FastAPI() + +oauth = OAuth( + framework="fastapi", + app=app +) + +@app.get("/login") +async def login(request: Request): + """Redirect to Kinde login page.""" + url = await oauth.login() + return RedirectResponse(url=url) + +@app.get("/register") +async def register(request: Request): + """Redirect to Kinde registration page.""" + url = await oauth.register() + return RedirectResponse(url=url) + +@app.get("/callback") +async def callback(request: Request, code: str, state: str = None): + """Handle the OAuth callback from Kinde.""" + try: + result = await oauth.handle_redirect(code, state) + return RedirectResponse(url="/") + except Exception as e: + raise HTTPException(status_code=400, detail=f"Authentication failed: {str(e)}") + +@app.get("/logout") +async def logout(request: Request): + """Logout the user and redirect to Kinde logout page.""" + await oauth.logout() + return RedirectResponse(url="/") + +@app.get("/user") +async def get_user(request: Request): + """Get the current user's information.""" + if not await oauth.is_authenticated(request): + url = await oauth.login() + return RedirectResponse(url=url) + return await oauth.get_user_info(request) +``` + +**Using AsyncOAuth client with FastAPI (for more control):** + +```python +from fastapi import FastAPI, Request, HTTPException +from fastapi.responses import RedirectResponse +from kinde_sdk.auth.async_oauth import AsyncOAuth + +app = FastAPI() +oauth = AsyncOAuth() + +@app.get("/login") +async def login(request: Request): + url = await oauth.login() + return RedirectResponse(url=url) + +@app.get("/callback") +async def callback(request: Request, code: str, state: str = None): + try: + result = await oauth.handle_redirect(code, state) + return RedirectResponse(url="/") + except Exception as e: + raise HTTPException(status_code=400, detail=str(e)) +``` + ### Manual route implementation If you prefer to implement the routes manually, here's how you can do it: @@ -305,54 +718,167 @@ async def get_user(request: Request): 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. +### Authentication flow steps + +1. **User initiates login/registration**: User clicks login/register button +2. **Redirect to Kinde**: Application redirects user to Kinde's authorization server +3. **User authenticates**: User signs in/up on Kinde's hosted pages +4. **Callback handling**: Kinde redirects back to your callback URL with authorization code +5. **Token exchange**: SDK exchanges authorization code for access tokens +6. **Session creation**: SDK stores tokens and creates user session +7. **Protected resources**: User can now access protected routes and resources + ## User permissions -The Kinde Python SDK provides a simple way to check user permissions in your application. First, import the permissions module: +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 import permissions -``` +from kinde_sdk.auth.oauth import OAuth +from flask import request +import asyncio -### Checking permissions +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... +``` -To check if a user has a specific permission: +### With AsyncOAuth client (Native async) ```python -# Check a single permission -permission = await permissions.get_permission("create:todos") -if permission["isGranted"]: - # User has permission - print(f"User has permission in organization: {permission['orgCode']}") +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 ``` -To get all permissions for the current user: +### With SmartOAuth client (Context-aware) ```python -# Get all permissions -all_permissions = await permissions.get_permissions() -print(f"User belongs to organization: {all_permissions['orgCode']}") -print("User permissions:", all_permissions["permissions"]) +from kinde_sdk.auth.smart_oauth import SmartOAuth + +oauth = SmartOAuth() + +# 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"] ``` +### Checking permissions + ### Practical examples -Here's how to use permissions in your application: +Here's how to use permissions in your application with different client types: + +**Example 1: Permission check in FastAPI (OAuth client)** ```python -# Example 1: Conditional Feature Access -async def create_todo_button(): - permission = await permissions.get_permission("create:todos") - if permission["isGranted"]: - return " - return None +from fastapi import FastAPI, Request, HTTPException +from kinde_sdk.auth.oauth import OAuth + +app = FastAPI() +oauth = OAuth(framework="fastapi", app=app) -# Example 2: Permission-Based API Endpoint -@router.post("/todos") -async def create_todo(todo_data: dict): - permission = await permissions.get_permission("create:todos") +@app.post("/todos") +async def create_todo(request: Request, todo_data: dict): + permission = await oauth.get_permission("create:todos", request) 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) + 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 3: Permission-based conditional rendering (Flask)** + +```python +from flask import Flask, request, render_template +from kinde_sdk.auth.oauth import OAuth +import asyncio + +app = Flask(__name__) +oauth = OAuth(framework="flask", app=app) + +@app.route("/dashboard") +def dashboard(): + loop = asyncio.get_event_loop() + can_create = loop.run_until_complete( + oauth.get_permission("create:todos", request) + )["isGranted"] + + can_delete = loop.run_until_complete( + oauth.get_permission("delete:todos", request) + )["isGranted"] + + return render_template("dashboard.html", + can_create=can_create, + can_delete=can_delete) ``` ### Common permission patterns @@ -381,73 +907,157 @@ For more information about setting up permissions in Kinde, see [User permission ## Feature flags -The Kinde Python SDK provides a simple way to access feature flags from your application. First, import the feature flags module: +The Kinde Python SDK provides a simple way to access feature flags from your application. Feature flags support both sync and async patterns. + +### With OAuth client (Framework-based) ```python -from kinde_sdk.auth import feature_flags -``` +from kinde_sdk.auth.oauth import OAuth +from flask import request +import asyncio -### Getting feature flags +oauth = OAuth(framework="flask", app=app) -To get a specific feature 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 -# Get a string feature flag -theme_flag = await feature_flags.get_flag("theme") -print(f"Current theme: {theme_flag.value}") +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") -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 + } ``` -To get all feature flags for the current user: +### Getting feature flags -```python -# 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})") -``` +To get a specific feature flag value: ### Practical examples -Here's how to use feature flags in your application: +Here's how to use feature flags in your application with different client types: + +**Example 1: Conditional feature rendering (FastAPI with OAuth)** ```python -# Example 1: Conditional Feature Rendering -async def render_create_competition_button(): - can_create = await feature_flags.get_flag("create_competition", default_value=False) +from fastapi import FastAPI, Request +from kinde_sdk.auth.oauth import OAuth + +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) if can_create.value: - return " - return None + return {"html": ""} + return {"html": ""} +``` + +**Example 2: Theme configuration (AsyncOAuth)** + +```python +from kinde_sdk.auth.async_oauth import AsyncOAuth + +oauth = AsyncOAuth() -# Example 2: Theme Configuration async def get_user_theme(): - theme = await feature_flags.get_flag("theme", default_value="light") - dark_mode = await feature_flags.get_flag("is_dark_mode", default_value=False) + theme = await oauth.get_flag("theme", default_value="light") + dark_mode = await oauth.get_flag("is_dark_mode", default_value=False) return { "theme": theme.value, "is_dark_mode": dark_mode.value } +``` + +**Example 3: Feature limits with validation (FastAPI)** -# Example 3: Feature Limits -@router.post("/competitions") -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() +```python +from fastapi import FastAPI, Request, HTTPException +from kinde_sdk.auth.oauth import OAuth + +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) if current_count >= limit_flag.value: raise HTTPException( status_code=403, - detail=f"Competition limit reached (max: {limit_flag.value}) + detail=f"Competition limit reached (max: {limit_flag.value})" ) # Create competition logic here... + return {"message": "Competition created"} +``` + +**Example 4: Type-safe flag access (AsyncOAuth)** + +```python +from kinde_sdk.auth.async_oauth import AsyncOAuth + +oauth = AsyncOAuth() + +async def get_all_flags(): + # Get all feature flags + all_flags = await oauth.get_all_flags() + result = {} + for code, flag in all_flags.items(): + result[code] = { + "value": flag.value, + "type": flag.type, + "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 @@ -493,62 +1103,154 @@ 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. First, import the claims module: - -```python -from kinde_sdk.auth import claims -``` - -### Getting claims +The Kinde Python SDK provides a simple way to access user claims from your application. Claims support both sync and async patterns. -To get a specific claim from the user's tokens: +### With OAuth client (Framework-based) ```python -# Get the audience claim from the access token -claim = await claims.get_claim("aud") -print(f"Token audience: {claim['value']}") +from kinde_sdk.auth.oauth import OAuth +from flask import request +import asyncio -# Get the given_name claim from the ID token -claim = await claims.get_claim("given_name", token_type="id_token") -print(f"User's given name: {claim['value']}") +oauth = OAuth(framework="flask", app=app) + +# 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"] + +# 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"] + } ``` -To get all claims from the user's tokens: +### With AsyncOAuth client (Native async) ```python -# Get all claims from the access token -all_claims = await claims.get_all_claims() -for claim_name, claim_value in all_claims.items(): - print(f"- {claim_name}: {claim_value}") +from kinde_sdk.auth.async_oauth import AsyncOAuth -# Get all claims from the ID token -id_token_claims = await claims.get_all_claims(token_type="id_token") +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: + ### Practical examples -Here's how to use claims in your application: +Here's how to use claims in your application with different client types: + +**Example 1: Accessing user information (FastAPI with OAuth)** ```python -# Example 1: Accessing User Information -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") +from fastapi import FastAPI, Request +from kinde_sdk.auth.oauth import OAuth + +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") if given_name["value"] and family_name["value"]: return { "name": f"{given_name['value']} {family_name['value']}", - "email": (await claims.get_claim("email", token_type="id_token"))["value"] + "email": email["value"] } return None +``` -# Example 2: Token Validation -@router.get("/api/protected") -async def protected_endpoint(): - aud_claim = await claims.get_claim("aud") +**Example 2: Token validation (AsyncOAuth)** + +```python +from fastapi import HTTPException +from kinde_sdk.auth.async_oauth import AsyncOAuth + +oauth = AsyncOAuth() + +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.get("/api/protected") +async def protected_endpoint(): + return await validate_token() +``` + +**Example 3: Getting all claims (AsyncOAuth)** + +```python +from kinde_sdk.auth.async_oauth import AsyncOAuth + +oauth = AsyncOAuth() + +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 + + # 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 + + return { + "access_token": access_token_data, + "id_token": id_token_data + } +``` + +**Example 4: Organization context (FastAPI)** + +```python +from fastapi import FastAPI, Request +from kinde_sdk.auth.oauth import OAuth + +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) + + return { + "org_code": org_code["value"], + "org_name": org_name["value"] if org_name else None + } ``` ### Common claims @@ -684,86 +1386,216 @@ 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 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. ### Getting started -To use the Management API, you'll need to initialize the client with your Kinde credentials: +#### With OAuth client (Framework-based) ```python from kinde_sdk.auth.oauth import OAuth +from flask import Flask +import asyncio -oauth = OAuth( - framework="flask", - app=app -) +app = Flask(__name__) +oauth = OAuth(framework="flask", app=app) # Get the management client management = oauth.get_management() + +# Use with asyncio in Flask +def list_users_sync(): + loop = asyncio.get_event_loop() + users = loop.run_until_complete(management.get_users()) + return users +``` + +#### With AsyncOAuth client (Native async) + +```python +from kinde_sdk.auth.async_oauth import AsyncOAuth + +oauth = AsyncOAuth() + +# Get the management client (native async) +management = await oauth.get_management() + +# All methods are async +users = await management.get_users() +``` + +#### With SmartOAuth client (Context-aware) + +```python +from kinde_sdk.auth.smart_oauth import SmartOAuth + +oauth = SmartOAuth() + +# 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() ``` ### Available endpoints -The Management API provides methods for common operations on resources. Here are some examples: +The Management API provides methods for common operations on resources. All examples use async patterns: + +**User management:** ```python -# List users +# List users (async) users = await management.get_users() -# Get a specific user +# Get a specific user (async) user = await management.get_user(user_id="user_123") -# Create a new user +# Create a new user (async) new_user = await management.create_user( email="user@example.com", given_name="John", family_name="Doe" ) -# Update a user +# Update a user (async) updated_user = await management.update_user( user_id="user_123", given_name="Johnny" ) -# Delete a user +# Delete a user (async) await management.delete_user(user_id="user_123") ``` ### Organization management +**Using Management API with FastAPI (OAuth client):** + ```python -# List organizations -orgs = await management.get_organizations() +from fastapi import FastAPI, HTTPException +from kinde_sdk.auth.oauth import OAuth -# Get a specific organization -org = await management.get_organization(org_id="org_123") +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 + +@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"} +``` -# Create a new organization -new_org = await management.create_organization( - name="My Organization" -) +**Using Management API with AsyncOAuth client:** -# Update an organization -updated_org = await management.update_organization( - org_id="org_123", - name="Updated Name" -) +```python +from kinde_sdk.auth.async_oauth import AsyncOAuth + +oauth = AsyncOAuth() -# Delete an organization -await management.delete_organization(org_id="org_123") +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 ``` ### Error handling The Management API methods will raise exceptions for API errors. It's recommended to handle these appropriately: +**Example with OAuth client (FastAPI):** + +```python +from fastapi import FastAPI, HTTPException +from kinde_sdk.auth.oauth import OAuth +from kinde_sdk.exceptions import KindeAPIException + +app = FastAPI() +oauth = OAuth(framework="fastapi", app=app) + +@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 -try: - user = await management.get_user(user_id="user_123") -except Exception as e: - # Handle API-specific errors - print(f"Error: {e}") +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 ``` ### Token management @@ -779,11 +1611,224 @@ You don't need to manually manage Management API tokens - the client handles thi ### Best practices -1. Always use async/await when calling Management API methods -2. Handle API errors appropriately -3. Cache results when appropriate to reduce API calls -4. Use appropriate error handling for production environments -5. Keep your client credentials secure +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 For more information about the Management API endpoints and capabilities, see the [Kinde Management API documentation](https://docs.kinde.com/kinde-apis/management/). +## Error handling and best practices + +Proper error handling is crucial for building robust applications with the Kinde Python SDK. This section covers common error scenarios and best practices. + +### Common exceptions + +The SDK raises specific exception types that you should handle: + +```python +from kinde_sdk.exceptions import ( + KindeAPIException, + KindeAuthenticationException, + KindeAuthorizationException, + KindeValidationException, + KindeConfigurationException +) +``` + +### Error handling patterns + +**Pattern 1: Comprehensive error handling (FastAPI)** + +```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)}") +``` + +**Pattern 2: Error handling with AsyncOAuth** + +```python +from kinde_sdk.auth.async_oauth import AsyncOAuth +from kinde_sdk.exceptions import KindeAPIException +import logging + +oauth = AsyncOAuth() +logger = logging.getLogger(__name__) + +async def safe_get_user(): + 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 None + except Exception as e: + logger.exception(f"Unexpected error: {str(e)}") + return None +``` + +**Pattern 3: Permission checking with error handling** + +```python +from fastapi import HTTPException +from kinde_sdk.auth.oauth import OAuth +from kinde_sdk.exceptions import KindeAPIException + +oauth = OAuth(framework="fastapi", app=app) + +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: + raise HTTPException( + status_code=e.status_code, + detail=f"Error checking permission: {str(e)}" + ) +``` + +### Best practices summary + +#### Authentication and authorization + +1. **Always validate authentication**: Check `is_authenticated()` before accessing protected resources +2. **Handle token expiration**: Implement token refresh logic or redirect to login +3. **Validate permissions**: Check permissions before allowing operations +4. **Use appropriate HTTP status codes**: 401 for authentication, 403 for authorization failures + +#### Configuration + +1. **Use environment variables**: Never hardcode credentials +2. **Validate configuration**: Check required environment variables at startup +3. **Use secure storage**: Store tokens securely (encrypted sessions, secure cookies) +4. **Set appropriate session timeouts**: Balance security and user experience + +#### Performance + +1. **Cache when appropriate**: Cache user info, permissions, and feature flags +2. **Use async operations**: Prefer async/await for better performance +3. **Implement connection pooling**: For high-traffic applications +4. **Batch operations**: When possible, batch Management API calls + +#### Security + +1. **Protect client secrets**: Use environment variables or secret management services +2. **Use HTTPS**: Always use HTTPS in production +3. **Validate redirect URLs**: Ensure callback URLs are whitelisted +4. **Implement CSRF protection**: Use state parameters in OAuth flows +5. **Log security events**: Log authentication failures and authorization denials + +#### Error handling + +1. **Handle specific exceptions**: Catch specific exception types, not just `Exception` +2. **Log errors appropriately**: Log errors with context but don't expose sensitive information +3. **Provide user-friendly messages**: Don't expose internal error details to users +4. **Implement retry logic**: For transient failures, implement exponential backoff +5. **Monitor error rates**: Track error rates and patterns in production + +#### Development + +1. **Use type hints**: Improve code clarity and IDE support +2. **Write tests**: Test authentication flows, error cases, and edge cases +3. **Use linting**: Catch errors early with tools like `mypy` and `ruff` +4. **Document error handling**: Document how your application handles errors + +### Example: Complete error handling setup + +```python +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 +) +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)}") + return JSONResponse( + status_code=401, + content={"error": "Authentication required", "detail": "Please log in"} + ) + +@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.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"} + ) + +@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"} +```