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"} ```