### Install frameio-kit and uvicorn using pip or uv Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/getting_started.md Installs the required Python packages for the Frame.io integration. Supports both uv and pip package managers. Run these commands in your project directory to set up dependencies. ```bash # Create project directory mkdir my-frameio-app cd my-frameio-app # Install frameio-kit (we recommend uv for fast, reliable installs) uv add frameio-kit uvicorn # Or with pip pip install frameio-kit uvicorn ``` -------------------------------- ### Start the Frame.io integration server using uvicorn Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/getting_started.md Launches the FastAPI/Starlette app defined in main.py with hot reload for development. ```bash uvicorn main:app --reload ``` -------------------------------- ### Define Frame.io integration app with custom action and webhook Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/getting_started.md Python example that creates an App instance, registers a custom action responding to user clicks, and a webhook handling file.ready events. Uses frameio_kit classes and async functions. ```python import os from frameio_kit import App, ActionEvent, WebhookEvent, Message app = App() # Custom Action: Responds to user clicks in Frame.io @app.on_action( event_type="greeting.say_hello", name="Say Hello", description="A simple greeting action", secret=os.environ["ACTION_SECRET"] ) async def on_greeting(event: ActionEvent): print(f"Hello from {event.user.id}!") return Message( title="Greetings!", description="Hello from your first frameio-kit app!" ) # Webhook: Responds to file events from Frame.io @app.on_webhook("file.ready", secret=os.environ["WEBHOOK_SECRET"]) async def on_file_ready(event: WebhookEvent): print(f"File {event.resource_id} is ready!") ``` -------------------------------- ### Expose local server to the internet with ngrok Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/getting_started.md Runs ngrok to forward port 8000, providing a public HTTPS URL for Frame.io to reach your webhook and action endpoints. ```bash ngrok http 8000 ``` -------------------------------- ### Bash: Development Environment Setup Commands Source: https://github.com/billyshambrook/frameio-kit/blob/main/README.md Commands for setting up the development environment after cloning the repository. Includes syncing dependencies with uv, activating the virtual environment, installing pre-commit hooks, and running development tasks like tests, code quality checks, and documentation serving. ```bash git clone https://github.com/billyshambrook/frameio-kit.git cd frameio-kit ``` ```bash uv sync source .venv/bin/activate ``` ```bash uv run pre-commit install ``` ```bash uv run pytest ``` ```bash uv run pre-commit run --all-files ``` ```bash uv run mkdocs serve ``` -------------------------------- ### Configure environment variables for Action and Webhook secrets Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/getting_started.md Stores signing secrets required for verifying requests from Frame.io. Place these lines in a .env file and load them when running the application. ```bash ACTION_SECRET=your-action-secret-here WEBHOOK_SECRET=your-webhook-secret-here ``` -------------------------------- ### Docker deployment configuration Source: https://context7.com/billyshambrook/frameio-kit/llms.txt Dockerfile configuration for deploying the application with uvicorn. Copies application files and installs Python dependencies before starting the server. ```dockerfile FROM python:3.12-slim WORKDIR /app COPY . . RUN pip install -r requirements.txt CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"] ``` -------------------------------- ### Bash: Frame.io Kit Installation Commands Source: https://github.com/billyshambrook/frameio-kit/blob/main/README.md Shows two methods for installing frameio-kit: using uv (recommended) for fast, reliable installs, or pip as an alternative. Installation requires Python 3.13+ and can be done with either modern uv package manager or traditional pip. ```bash uv add frameio-kit ``` ```bash pip install frameio-kit ``` -------------------------------- ### Configure Token Storage Backends - Python Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/user_auth.md Examples of different token storage backends for various deployment scenarios. MemoryStore for development (default), DiskStore for single servers, and RedisStore for multi-server deployments. ```python # Development: MemoryStore (default - no storage parameter needed) app = App(oauth=OAuthConfig(...)) # Single Server: DiskStore from key_value.aio.stores.disk import DiskStore app = App( oauth=OAuthConfig( ..., storage=DiskStore(directory="./tokens"), ) ) # Multi-Server: RedisStore from key_value.aio.stores.redis import RedisStore app = App( oauth=OAuthConfig( ..., storage=RedisStore(url="redis://localhost:6379"), ) ) ``` -------------------------------- ### Python API Client Structure Example Source: https://github.com/billyshambrook/frameio-kit/blob/main/AGENT.md Illustrates the recommended structure for the Frameio API client, organizing client methods by API resources (e.g., FilesClient, CommentsClient). This mirrors the REST API structure for better intuitiveness, contrasting with a less organized flat structure. ```python # Good: Intuitive and mirrors the REST API class FrameioClient: def __init__(self, token: str): self.files = FilesClient(...) self.comments = CommentsClient(...) ``` ```python # Bad: A flat client with no organization class FrameioClient: async def get_file(...) -> ...: ... async def create_comment(...) -> ...: ... ``` -------------------------------- ### Implementing TimingMiddleware with __call__ in Python Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/middleware.md This example creates a TimingMiddleware to measure event processing duration using time.monotonic(). Runs universally via __call__; prints completion time. Depends on frameio_kit and time; integrates with App setup and webhook handlers. Limitation: console output only, no metrics export. ```python import time from frameio_kit import App, Middleware, WebhookEvent, Message from frameio_kit.middleware import AnyEvent, NextFunc, AnyResponse class TimingMiddleware(Middleware): async def __call__(self, event: AnyEvent, next: NextFunc) -> AnyResponse: start_time = time.monotonic() try: return await next(event) finally: duration = time.monotonic() - start_time print(f"Completed in {duration:.2f}s") # Usage app = App( token=os.getenv("FRAMEIO_TOKEN"), middleware=[TimingMiddleware()] ) @app.on_webhook("file.ready", secret=os.environ["WEBHOOK_SECRET"]) async def on_file_ready(event: WebhookEvent): print("File ready") ``` -------------------------------- ### Python Docstring Example for get_file Method Source: https://github.com/billyshambrook/frameio-kit/blob/main/AGENT.md Demonstrates the use of Google-style docstrings for an asynchronous Python method. It includes details about the method's purpose, API endpoint, arguments, return type, and potential exceptions. This follows the project's requirement for documenting public methods. ```python async def get_file(self, file_id: str) -> File: """Retrieves a specific file by its unique ID. This function makes an asynchronous call to the `GET /files/{file_id}` endpoint. Args: file_id: The unique identifier for the file. Returns: A Pydantic model of the requested File. Raises: FrameIOApiException: If the API returns a non-2xx status code. """ response = await self.http.get(f"files/{file_id}") response.raise_for_status() # Let httpx handle raising exceptions for bad statuses return File(**response.json()) ``` -------------------------------- ### Python: Frame.io App Setup and Event Handling Source: https://github.com/billyshambrook/frameio-kit/blob/main/README.md Demonstrates how to create a Frame.io integration app using frameio-kit. Shows app initialization, webhook event handling for file.ready events with secret validation, and custom action creation with message responses. Uses async/await pattern and decorator-based routing for clean event management. ```python import os from frameio_kit import App, WebhookEvent, ActionEvent, Message app = App() @app.on_webhook("file.ready", secret=os.environ["WEBHOOK_SECRET"]) async def on_file_ready(event: WebhookEvent): print(f"File {event.resource_id} is ready!") @app.on_action("my_app.analyze", "Analyze File", "Analyze this file", os.environ["ACTION_SECRET"]) async def analyze_file(event: ActionEvent): return Message(title="Analysis Complete", description="File analyzed successfully!") ``` -------------------------------- ### User Authentication with OAuth Token Client (Python) Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/client_api.md This snippet creates a user-specific client using an OAuth token for action events, attributing API calls to the user, and shows fetching a file. Dependencies: frameio_kit (Client, get_user_token, ActionEvent). Inputs: Action event with require_user_auth=True; outputs: User-attributed file data. Limitations: Requires user auth setup; client must be closed; see user auth guide for OAuth details. ```python from frameio_kit import Client, get_user_token @app.on_action(..., require_user_auth=True) async def my_action(event: ActionEvent): # Create client with user's token from context user_client = Client(token=get_user_token()) # API calls are now attributed to the user file = await user_client.files.show(...) await user_client.close() ``` -------------------------------- ### Complete OAuth Setup with Redis Storage - Python Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/user_auth.md Comprehensive OAuth configuration with custom scopes, Redis storage backend, and explicit encryption key. Suitable for production deployments requiring persistent token storage across multiple servers. ```python from key_value.aio.stores.redis import RedisStore app = App( oauth=OAuthConfig( client_id=os.environ["ADOBE_CLIENT_ID"], client_secret=os.environ["ADOBE_CLIENT_SECRET"], base_url="https://yourapp.com", scopes=["openid", "AdobeID", "frameio.api"], storage=RedisStore(url="redis://localhost:6379"), encryption_key=os.environ["FRAMEIO_AUTH_ENCRYPTION_KEY"], ) ) ``` -------------------------------- ### Full Example: Notification Action with Message Response in Python Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/custom_actions.md Combines action registration, event handling, and a Message response to send a notification about an asset. Uses environment variables for the secret and demonstrates async notification logic. ```python import os from frameio_kit import App, ActionEvent, Message app = App() @app.on_action( event_type="asset.notify", name="Notify Team", description="Send notification about this asset", secret=os.environ["ACTION_SECRET"] ) async def notify_team(event: ActionEvent): print(f"Notification sent for {event.resource_id} by {event.user.name}") # Send actual notification here await send_notification(event.resource_id, event.user.id) return Message( title="Notification Sent", description="Your team has been notified about this asset." ) ``` -------------------------------- ### Complete Frame.io Python Production Integration Source: https://context7.com/billyshambrook/frameio-kit/llms.txt Full production setup for Frame.io application with middleware, OAuth configuration using Redis storage, webhook handlers for file events (ready, deleted, trashed), and proper lifecycle management. Includes request logging middleware, error handling, and integration with Redis for state persistence. ```python import os import logging from typing import AsyncIterator from contextlib import asynccontextmanager from frameio_kit import ( App, OAuthConfig, WebhookEvent, ActionEvent, Message, Form, TextField, SelectField, SelectOption, Middleware, NextFunc, AnyEvent, AnyResponse, Client, get_user_token ) from key_value.aio.stores.redis import RedisStore # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Custom middleware class RequestLoggingMiddleware(Middleware): async def __call__(self, event: AnyEvent, next: NextFunc) -> AnyResponse: logger.info(f"Processing {event.type} event") try: response = await next(event) logger.info(f"Successfully processed {event.type}") return response except Exception as e: logger.error(f"Error processing {event.type}: {e}", exc_info=True) raise # Lifespan context manager for startup/shutdown @asynccontextmanager async def lifespan(app: App) -> AsyncIterator[None]: # Startup logger.info("Starting Frame.io integration") await initialize_services() yield # Shutdown logger.info("Shutting down Frame.io integration") if app.client: await app.client.close() if hasattr(app, '_oauth_client') and app._oauth_client: await app._oauth_client.close() # Initialize app with all features app = App( token=os.environ["FRAMEIO_TOKEN"], middleware=[RequestLoggingMiddleware()], oauth=OAuthConfig( client_id=os.environ["ADOBE_CLIENT_ID"], client_secret=os.environ["ADOBE_CLIENT_SECRET"], base_url=os.environ["APP_BASE_URL"], scopes=["openid", "creative_sdk"], storage=RedisStore(url=os.environ["REDIS_URL"]), encryption_key=os.environ["FRAMEIO_AUTH_ENCRYPTION_KEY"], token_refresh_buffer_seconds=300 ) ) # Webhook handlers @app.on_webhook("file.ready", secret=os.environ["WEBHOOK_SECRET"]) async def on_file_ready(event: WebhookEvent): """Process newly uploaded files""" logger.info(f"File ready: {event.resource_id} in project {event.project_id}") # Get file details file = await app.client.files.show( account_id=event.account_id, file_id=event.resource_id ) # Trigger processing pipeline await process_media_file( file_id=event.resource_id, file_name=file.data.name, project_id=event.project_id, uploaded_by=event.user.name ) # Add comment to file await app.client.comments.create( account_id=event.account_id, asset_id=event.resource_id, text="File received and queued for processing" ) @app.on_webhook( event_type=["file.deleted", "file.trashed"], secret=os.environ["WEBHOOK_SECRET"] ) async def on_file_removed(event: WebhookEvent): """Cleanup when files are deleted""" logger.info(f"File removed: {event.resource_id}") await cleanup_processed_data(event.resource_id) ``` -------------------------------- ### Basic Frame.io Webhook and Action Setup in Python Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/index.md This snippet demonstrates initializing a Frame.io app and defining handlers for webhook events and custom actions using decorators. It requires the frameio_kit library and a secret for verification. Inputs include event types and secrets; outputs are async functions that process events, such as printing messages or returning responses. Limitations include needing Python 3.13+ and proper secret management. ```python from frameio_kit import App, WebhookEvent, ActionEvent, Message app = App() @app.on_webhook("file.ready", secret="your-secret") async def on_file_ready(event: WebhookEvent): print(f"File is ready!") @app.on_action("my_app.analyze", "Analyze File", "Analyze this file", "your-secret") async def analyze_file(event: ActionEvent): return Message(title="Analysis Complete", description="File analyzed successfully!") ``` -------------------------------- ### Create Conditional Middleware Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/middleware.md Implements middleware that executes only when a specific condition is met, determined by a function passed during initialization. Useful for applying logic selectively based on event properties. The example filters events starting with 'file.'. ```python class ConditionalMiddleware(Middleware): def __init__(self, condition_func): self.condition_func = condition_func async def __call__(self, event: AnyEvent, next: NextFunc) -> AnyResponse: if self.condition_func(event): # Only run middleware logic if condition is met print(f"Condition met for {event.type}") return await next(event) # Usage app = App( middleware=[ ConditionalMiddleware(lambda e: e.type.startswith("file.")), ] ) ``` -------------------------------- ### Initialize Frame.io App with Middleware Chain Source: https://context7.com/billyshambrook/frameio-kit/llms.txt App initialization with middleware chain configuration enables custom request processing. Multiple middleware components are registered in execution order, with each middleware wrapping subsequent handlers. The example shows Integration with TimingMiddleware, LoggingMiddleware, ErrorHandlingMiddleware, and MetricsMiddleware for comprehensive request handling and monitoring. ```python # Initialize app with middleware chain (executes in order) app = App( token=os.environ["FRAMEIO_TOKEN"], middleware=[ TimingMiddleware(), LoggingMiddleware(), ErrorHandlingMiddleware(), MetricsMiddleware(), ] ) @app.on_webhook("file.ready", secret=os.environ["WEBHOOK_SECRET"]) async def handle_file(event: WebhookEvent): """All middleware runs before and after this handler""" # Middleware chain: # 1. TimingMiddleware starts timer # 2. LoggingMiddleware logs event # 3. ErrorHandlingMiddleware wraps in try/catch # 4. MetricsMiddleware increments counter # 5. Handler executes # 6. Middleware chain unwinds in reverse order await process_file(event.resource_id) ``` -------------------------------- ### Server-to-Server Authentication with API Client (Python) Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/client_api.md This example initializes the App for server-to-server authentication and makes a file show call using the client. Dependencies: frameio_kit App and os for token. Inputs: FRAMEIO_TOKEN env var; outputs: File data from API. Limitations: Token handles auth automatically but needs proper permissions; all calls are async. ```python app = App(token=os.getenv("FRAMEIO_TOKEN")) # Use app.client for S2S authenticated calls file = await app.client.files.show(...) ``` -------------------------------- ### Frame.io API Client - Authenticated Operations Source: https://context7.com/billyshambrook/frameio-kit/llms.txt Demonstrates Frame.io REST API client usage with server-to-server token authentication. Shows comprehensive endpoint operations including file details, project information, comment management, and team member access. Includes both app-based and standalone client examples with proper resource management. ```python import os from frameio_kit import App, Client, ActionEvent, Message, get_user_token app = App(token=os.environ["FRAMEIO_TOKEN"]) @app.on_action( event_type="myapp.analyze", name="Analyze Asset", description="Get detailed asset information", secret=os.environ["ACTION_SECRET"] ) async def analyze_asset(event: ActionEvent): """Use API client to fetch and analyze asset data""" # Access app's client (uses S2S token) client = app.client # Get file details file = await client.files.show( account_id=event.account_id, file_id=event.resource_id ) # Get project details project = await client.projects.show( account_id=event.account_id, project_id=event.project_id ) # List comments on the file comments = await client.comments.list( account_id=event.account_id, asset_id=event.resource_id ) # Create a new comment await client.comments.create( account_id=event.account_id, asset_id=event.resource_id, text=f"Analyzed by {event.user.name}", timestamp=0 # Optional frame number ) # Get team members team = await client.teams.list(account_id=event.account_id) analysis = f""" File: {file.data.name} ({file.data.filesize} bytes) Project: {project.data.name} Comments: {len(comments.data)} Team Members: {len(team.data)} """ return Message( title="Analysis Complete", description=analysis ) # Create standalone client async def standalone_example(): """Use client outside of app context""" client = Client( token=os.environ["FRAMEIO_TOKEN"], timeout=30.0, follow_redirects=True ) try: # Get current user profile profile = await client.users.show() print(f"Logged in as: {profile.data.name}") # List all accounts accounts = await client.accounts.list() for account in accounts.data: # List projects in each account projects = await client.projects.list(account_id=account.id) print(f"Account {account.name}: {len(projects.data)} projects") finally: # Always close client when done await client.close() ``` -------------------------------- ### Process File Ready Event and Trigger Custom Logic (Python) Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/webhooks.md Shows a complete handler for the file.ready event, printing a message and calling an async processing function. Includes environment variable for secret and illustrates an async workflow inside a webhook. ```Python import os from frameio_kit import App, WebhookEvent, Message app = App() @app.on_webhook("file.ready", secret=os.environ["WEBHOOK_SECRET"]) async def on_file_ready(event: WebhookEvent): print(f"File {event.resource_id} is ready for processing") # Simulate some processing await process_file(event.resource_id) async def process_file(file_id: str): # Your processing logic here pass ``` -------------------------------- ### OAuth Configuration - User Authentication Source: https://context7.com/billyshambrook/frameio-kit/llms.txt Configures Adobe Login OAuth integration for Frame.io user-specific API calls. Includes token encryption, storage backend setup, and automatic route generation. Supports custom HTTP clients and configurable token refresh buffering for secure user authentication flows. ```python import os from frameio_kit import App, OAuthConfig, ActionEvent, Message, Client, get_user_token from frameio_kit import TokenEncryption from key_value.aio.stores.redis import RedisStore # Generate encryption key (do this once, store securely) encryption_key = TokenEncryption.generate_key() print(f"Generated key: {encryption_key}") # Store this in environment variable: FRAMEIO_AUTH_ENCRYPTION_KEY # Initialize app with OAuth app = App( oauth=OAuthConfig( client_id=os.environ["ADOBE_CLIENT_ID"], client_secret=os.environ["ADOBE_CLIENT_SECRET"], base_url="https://myapp.com", # Your app's public URL scopes=["openid", "creative_sdk"], # Required OAuth scopes # Storage backend for tokens (supports multiple backends) storage=RedisStore(url="redis://localhost:6379"), # Encryption key for token storage encryption_key=os.environ["FRAMEIO_AUTH_ENCRYPTION_KEY"], # Token refresh buffer (default 300 seconds) token_refresh_buffer_seconds=300, # Optional custom HTTP client http_client=None ) ) # OAuth routes automatically added: # GET /auth/login?user_id={user_id} - Initiates login flow # GET /auth/callback - Handles OAuth callback ``` -------------------------------- ### Get metadata action in Python Source: https://context7.com/billyshambrook/frameio-kit/llms.txt Retrieves and displays detailed asset metadata using user authentication. Requires user token for API calls and ACTION_SECRET environment variable. Shows file details and requester information. ```python @app.on_action( event_type="myapp.get_metadata", name="Get Metadata", description="View detailed asset metadata", secret=os.environ["ACTION_SECRET"], require_user_auth=True ) async def get_metadata(event: ActionEvent): """Display metadata using user authentication""" # Use user's token for API calls user_token = get_user_token() user_client = Client(token=user_token) try: # Get file details as the user file = await user_client.files.show( account_id=event.account_id, file_id=event.resource_id ) # Get user profile profile = await user_client.users.show() # Format metadata metadata = f""" Filename: {file.data.name} Size: {file.data.filesize:,} bytes Type: {file.data.type} Created: {file.data.inserted_at} Requested by: {profile.data.name} """ return Message( title="Asset Metadata", description=metadata ) finally: await user_client.close() ``` -------------------------------- ### File Processing Pipeline with Frame.io Webhook in Python Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/index.md This example shows handling a 'file.ready' webhook to fetch file details, process the file externally, and add a comment back to Frame.io. Dependencies include frameio_kit and an external processing service. Inputs are webhook events with resource IDs; outputs include API calls for file retrieval and comment creation. Limitations involve async handling and environment variable for secrets. ```python @app.on_webhook("file.ready", secret=os.environ["WEBHOOK_SECRET"]) async def process_file(event: WebhookEvent): # Get file details file = await app.client.files.get(event.resource_id) # Process the file result = await my_processing_service.process(file) # Add a comment back to Frame.io await app.client.comments.create( account_id=event.account_id, file_id=event.resource_id, data={"text": f"Processing complete! Result: {result}"} ) ``` -------------------------------- ### Process File and Add Comment on Webhook Event (Python) Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/client_api.md This example handles a 'file.ready' webhook event by fetching file details, processing content, and creating a comment using the API client. Dependencies: frameio_kit (App, WebhookEvent), frameio (CreateCommentParamsData), and os for secrets. Inputs: Webhook event with account_id and resource_id; outputs: Printed file name and added comment. Limitations: Assumes async environment; requires valid token and webhook secret; processing logic is placeholder. ```python import os from frameio_kit import App, WebhookEvent, Message from frameio import CreateCommentParamsData app = App(token=os.getenv("FRAMEIO_TOKEN")) @app.on_webhook("file.ready", secret=os.environ["WEBHOOK_SECRET"]) async def process_file(event: WebhookEvent): # Get file details file = await app.client.files.show( account_id=event.account_id, file_id=event.resource_id ) print(f"Processing: {file.data.name}") # Simulate processing await process_file_content(file) # Add a comment to the file await app.client.comments.create( account_id=event.account_id, file_id=event.resource_id, data=CreateCommentParamsData(text="✅ File processed successfully!") ) async def process_file_content(file): # Your processing logic here pass ``` -------------------------------- ### Setting Up App with Middleware Instances in Python Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/middleware.md Create an App instance by passing middleware list for global application. Supports instances like LoggingMiddleware; requires FRAMEIO_TOKEN env var. Inputs: token and middleware list; outputs configured App. Applies to all handlers without per-handler duplication. ```python from frameio_kit import App, Middleware app = App( token=os.getenv("FRAMEIO_TOKEN"), middleware=[ LoggingMiddleware(), TimingMiddleware(), # Add more middleware here ] ) ``` -------------------------------- ### Initialize Frame.io App with Middleware and OAuth Source: https://context7.com/billyshambrook/frameio-kit/llms.txt Creates an instance of the App class with optional server-to-server token, middleware, and OAuth configuration. Supports webhook and action decorators for routing with built-in security. ```python import os from frameio_kit import App, WebhookEvent, ActionEvent, Message # Initialize app with server-to-server token and optional middleware app = App( token=os.environ["FRAMEIO_TOKEN"], # Optional S2S API token middleware=[], # Optional middleware list oauth=None # Optional OAuth configuration ) # Register webhook handler for file ready events @app.on_webhook("file.ready", secret=os.environ["WEBHOOK_SECRET"]) async def on_file_ready(event: WebhookEvent): """Called when a file is ready for processing""" print(f"File {event.resource_id} ready in project {event.project_id}") print(f"Uploaded by: {event.user.name}") # Access authenticated API client file = await app.client.files.show( account_id=event.account_id, file_id=event.resource_id ) print(f"File name: {file.data.name}, size: {file.data.filesize}") # Register custom action with multiple event types @app.on_action( event_type="myapp.notify", name="Notify Team", description="Send notification about this asset", secret=os.environ["ACTION_SECRET"] ) async def notify_team(event: ActionEvent): """Custom action that appears in Frame.io UI""" await send_notification(event.resource_id, event.user.name) return Message( title="Success", description="Team has been notified" ) # Run with uvicorn or any ASGI server # uvicorn myapp:app --host 0.0.0.0 --port 8000 ``` -------------------------------- ### Bash: Git Pull Request Workflow Commands Source: https://github.com/billyshambrook/frameio-kit/blob/main/README.md Step-by-step Git workflow for contributing to the project. Includes forking the repository, creating feature branches, making changes with proper testing, ensuring code quality, committing changes, and opening pull requests. Follows standard open source contribution practices. ```bash git clone https://github.com/billyshambrook/frameio-kit.git cd frameio-kit ``` -------------------------------- ### Helper functions in Python Source: https://context7.com/billyshambrook/frameio-kit/llms.txt Collection of utility functions including service initialization, media processing, cleanup, and export job management. These functions support the main application actions. ```python async def initialize_services(): """Initialize external services on startup""" logger.info("Initializing services...") # Connect to databases, message queues, etc. async def process_media_file(file_id: str, file_name: str, project_id: str, uploaded_by: str): """Process media file in background""" logger.info(f"Processing {file_name} (ID: {file_id})") # Implement actual processing logic async def cleanup_processed_data(file_id: str): """Cleanup associated data when file is deleted""" logger.info(f"Cleaning up data for file {file_id}") # Implement cleanup logic async def start_export_job(resource_id: str, format_type: str, include_audio: bool, user: str) -> str: """Start background export job""" logger.info(f"Starting export: {format_type} for {resource_id}") return "job_12345" # Return job ID ``` -------------------------------- ### Add Comment to Frame.io Using API Client in Webhook (Python) Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/webhooks.md Uses the Frame.io API client within a webhook handler to create a comment on the processed file. Shows how to pass an authentication token and use CreateCommentParamsData for the comment payload. ```Python import os from frameio import CreateCommentParamsData from frameio_kit import App, WebhookEvent, Message app = App(token=os.getenv("FRAMEIO_TOKEN")) @app.on_webhook("file.ready", secret=os.environ["WEBHOOK_SECRET"]) async def add_processing_comment(event: WebhookEvent): # Use the API client to add a comment back to Frame.io await app.client.comments.create( account_id=event.account_id, file_id=event.resource_id, data=CreateCommentParamsData(text="File has been automatically processed!") ) ``` -------------------------------- ### Access Webhook Event Data via WebhookEvent Object (Python) Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/webhooks.md Demonstrates importing WebhookEvent and printing its typed attributes such as type, resource_id, and related IDs. Useful for extracting event details within a handler. No external dependencies beyond frameio_kit. ```Python from frameio_kit import WebhookEvent async def handler(event: WebhookEvent): print(event.type) # "file.ready" print(event.resource_id) # "abc123" print(event.account_id) # "acc_456" print(event.user_id) # "user_789" print(event.project_id) # "proj_101" print(event.workspace_id) # "ws_123" ``` -------------------------------- ### Register Webhook Handler with @app.on_webhook Decorator (Python) Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/webhooks.md This snippet shows how to use the @app.on_webhook decorator to register an asynchronous function that handles a specific Frame.io webhook event. It specifies the event_type and signing secret, and receives a WebhookEvent object for custom processing. ```Python @app.on_webhook(event_type="file.ready", secret="your-secret") async def on_file_ready(event: WebhookEvent): # Handle the event pass ``` -------------------------------- ### Handle Multiple Comment Events in a Single Webhook (Python) Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/webhooks.md Registers a webhook handler for both comment.created and comment.updated events, determines the action, logs the comment ID, and calls an async notification function. Demonstrates use of a list of event types and conditional logic. ```Python from frameio_kit import App, WebhookEvent app = App() @app.on_webhook( event_type=["comment.created", "comment.updated"], secret=os.environ["WEBHOOK_SECRET"] ) async def on_comment_change(event: WebhookEvent): action = "created" if event.type == "comment.created" else "updated" print(f"Comment {action}: {event.resource_id}") # Send notification to team await notify_team(event) ``` -------------------------------- ### Initialize Frame.io App with API Token (Python) Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/client_api.md This snippet initializes the Frame.io App instance using an API token from environment variables, enabling access to the API client for authenticated calls. Dependencies include the frameio_kit library and os module for secure token handling. Inputs are the FRAMEIO_TOKEN environment variable; outputs an App object with app.client property. Limitations: Requires a valid token with appropriate scopes; async methods must be awaited. ```python import os from frameio_kit import App # Initialize with API token app = App(token=os.getenv("FRAMEIO_TOKEN")) # Client is now available at app.client ``` -------------------------------- ### Initialize OAuth Configuration - Python Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/user_auth.md Sets up Adobe OAuth credentials for user authentication in Frame.io Kit. Requires client ID and secret from Adobe Developer Console. Tokens are stored in memory by default and lost on restart. ```python import os from frameio_kit import App, OAuthConfig app = App( oauth=OAuthConfig( client_id=os.environ["ADOBE_CLIENT_ID"], client_secret=os.environ["ADOBE_CLIENT_SECRET"], base_url="https://yourapp.com", ) ) ``` -------------------------------- ### Return a Simple Message Response in Python Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/custom_actions.md Demonstrates creating and returning a Message object for actions that complete immediately without further user input. The Message includes a title and description displayed to the user. ```python from frameio_kit import Message return Message( title="Task Complete", description="Your file has been processed successfully!" ) ``` -------------------------------- ### Configure DynamoDB Storage Backend - Python Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/user_auth.md AWS DynamoDB storage backend for multi-server deployments. Requires DynamoDB table with partition key 'key' and TTL attribute 'ttl' for automatic token cleanup. ```python from key_value.aio.stores.dynamodb import DynamoDBStore app = App( oauth=OAuthConfig( ..., storage=DynamoDBStore( table_name="frameio-oauth-tokens", region_name="us-east-1", ), ) ) ``` -------------------------------- ### Return a Form for User Input in Python Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/custom_actions.md Illustrates building a Form with a TextField to collect user preferences. The Form is returned to the UI, prompting the user to fill out the fields before the handler is called again. ```python from frameio_kit import Form, TextField return Form( title="Configure Settings", description="Enter your preferences:", fields=[ TextField(label="Name", name="name", value="default") ] ) ``` -------------------------------- ### Implement two-step form process pattern Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/custom_actions.md Demonstrates the two-step process pattern for custom actions with forms. First call shows the form (when event.data is None), second call processes the submitted data and returns a message. ```python async def my_action(event: ActionEvent): if event.data is None: # First call - show the form return Form(title="My Form", fields=[...]) else: # Second call - process the form data return Message(title="Done", description="Form processed!") ``` -------------------------------- ### Manually manage user tokens with TokenManager Source: https://context7.com/billyshambrook/frameio-kit/llms.txt Demonstrates how to manually store, retrieve, and delete user tokens using TokenManager. Shows how to create TokenData with expiration and scopes. Includes checking if a token is expired. ```python async def manual_token_example(): """Manually manage tokens using TokenManager""" token_manager = app.token_manager # Store token for a user from frameio_kit import TokenData from datetime import datetime, timedelta token_data = TokenData( access_token="access_token_here", refresh_token="refresh_token_here", expires_at=datetime.now() + timedelta(hours=1), scopes=["openid", "creative_sdk"], user_id="user123" ) await token_manager.store_token("user123", token_data) # Retrieve token retrieved = await token_manager.get_token("user123") if retrieved: print(f"Token expires: {retrieved.expires_at}") print(f"Is expired: {retrieved.is_expired()}") # Delete token await token_manager.delete_token("user123") ``` -------------------------------- ### OAuth Configuration - User Authentication Source: https://context7.com/billyshambrook/frameio-kit/llms.txt OAuth configuration enables Adobe Login integration for user-specific API calls with automatic token management and encryption. ```APIDOC ## OAuth Configuration - User Authentication ### Description OAuth configuration enables Adobe Login integration for user-specific API calls with automatic token management and encryption. ### Method GET ### Endpoint /auth/login?user_id={user_id} /auth/callback ### Parameters #### Query Parameters - **user_id** (string) - Required - The user ID ### Request Example { "client_id": "client_id", "client_secret": "client_secret", "base_url": "https://myapp.com", "scopes": ["openid", "creative_sdk"] } ### Response #### Success Response (200) - **token** (string) - The OAuth token #### Response Example { "token": "oauth_token" } ``` -------------------------------- ### Create Custom Action with Multiple UI Field Types Source: https://context7.com/billyshambrook/frameio-kit/llms.txt Demonstrates how to create a complex form with all available field types including text fields, textareas, select dropdowns, checkboxes, and link fields. Also shows simple message response implementation. ```python from frameio_kit import ( App, ActionEvent, Message, Form, TextField, TextareaField, SelectField, SelectOption, CheckboxField, LinkField ) app = App() # Simple message response @app.on_action( event_type="myapp.simple", name="Simple Action", description="Show a simple message", secret=os.environ["ACTION_SECRET"] ) async def simple_action(event: ActionEvent): """Return a message with title and description""" return Message( title="Action Complete", description="The operation was successful" ) # Complex form with all field types @app.on_action( event_type="myapp.complex_form", name="Complex Form", description="Show all field types", secret=os.environ["ACTION_SECRET"] ) async def complex_form_action(event: ActionEvent): """Demonstrate all available form field types""" if event.data: # Access submitted form data name = event.data["name"] bio = event.data["bio"] country = event.data["country"] subscribe = event.data["subscribe"] return Message( title="Form Submitted", description=f"Received data for {name} from {country}" ) # Show comprehensive form return Form( title="User Profile", description="Complete your profile information:", fields=[ # Single-line text input TextField( label="Full Name", name="name", value="John Doe" # Optional default value ), # Multi-line text input TextareaField( label="Biography", name="bio", value="Tell us about yourself..." ), # Dropdown select SelectField( label="Country", name="country", options=[ SelectOption(name="United States", value="us"), SelectOption(name="Canada", value="ca"), SelectOption(name="United Kingdom", value="uk"), ], value="us" # Optional default selection ), # Boolean checkbox CheckboxField( label="Subscribe to newsletter", name="subscribe", value=True # Optional default checked state ), # Display-only link with copy functionality LinkField( label="Profile URL", name="profile_url", value="https://example.com/profile/123" ) ] ) ``` -------------------------------- ### Implement Metrics Middleware for Webhook Tracking Source: https://context7.com/billyshambrook/frameio-kit/llms.txt The MetricsMiddleware class provides comprehensive tracking for webhook and action invocations with error counting. It increments counters for webhooks and actions, tracks exceptions, and exposes statistics through get_stats(). Compatible with Python async/await patterns and integrates seamlessly into the middleware chain. ```python class MetricsMiddleware(Middleware): """Track handler invocation counts and errors""" def __init__(self): self.webhook_count = 0 self.action_count = 0 self.error_count = 0 async def on_webhook(self, event: WebhookEvent, next: NextFunc) -> AnyResponse: self.webhook_count += 1 try: return await next(event) except Exception: self.error_count += 1 raise async def on_action(self, event: ActionEvent, next: NextFunc) -> AnyResponse: self.action_count += 1 try: return await next(event) except Exception: self.error_count += 1 raise def get_stats(self): return { "webhooks": self.webhook_count, "actions": self.action_count, "errors": self.error_count } ``` -------------------------------- ### Interactive Custom Action for Asset Publishing in Python Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/index.md This snippet implements a custom action for publishing assets to social media, including form display and submission handling. It uses frameio_kit for actions and requires platform options defined. Inputs are action events with optional form data; outputs are form responses or success messages after publishing. Limitations include needing to implement the external publish function and handling form validation. ```python @app.on_action("asset.publish", "Publish Asset", "Publish to social media", os.environ["ACTION_SECRET"]) async def publish_asset(event: ActionEvent): if event.data: # Form was submitted platform = event.data.get("platform") caption = event.data.get("caption") # Publish the asset await publish_to_social_media(event.resource_id, platform, caption) return Message(title="Published!", description=f"Posted to {platform}") # Show the form return Form( title="Publish to Social Media", fields=[ SelectField(label="Platform", name="platform", options=PLATFORMS), TextField(label="Caption", name="caption", placeholder="Enter your caption...") ] ) ``` -------------------------------- ### Log events with detailed information using middleware Source: https://context7.com/billyshambrook/frameio-kit/llms.txt Provides detailed logging for different event types using specialized handler methods. Logs webhook events with resource and user information, and action events with user and data status. ```python # Logging middleware with event-specific handlers class LoggingMiddleware(Middleware): """Log all incoming events with detailed information""" async def on_webhook(self, event: WebhookEvent, next: NextFunc) -> AnyResponse: logger.info( f"Webhook: {event.type} | " f"Resource: {event.resource_id} | " f"User: {event.user.name} | " f"Project: {event.project_id}" ) return await next(event) async def on_action(self, event: ActionEvent, next: NextFunc) -> AnyResponse: logger.info( f"Action: {event.type} | " f"User: {event.user.name} | " f"Resource: {event.resource_id} | " f"Has Data: {event.data is not None}" ) return await next(event) ``` -------------------------------- ### Implement Rate Limiting Middleware with State Source: https://github.com/billyshambrook/frameio-kit/blob/main/docs/usage/middleware.md Creates stateful middleware to track and limit the number of requests per resource within a time window. Maintains request timestamps in memory and raises an exception when limits are exceeded. Suitable for basic rate limiting scenarios. ```python class RateLimitMiddleware(Middleware): def __init__(self, max_requests_per_minute=60): self.max_requests = max_requests_per_minute self.requests = {} # Track requests by resource_id async def __call__(self, event: AnyEvent, next: NextFunc) -> AnyResponse: current_time = time.time() resource_id = event.resource_id # Clean old entries self.requests[resource_id] = [ req_time for req_time in self.requests.get(resource_id, []) if current_time - req_time < 60 ] # Check rate limit if len(self.requests.get(resource_id, [])) >= self.max_requests: raise Exception(f"Rate limit exceeded for resource {resource_id}") # Record this request self.requests.setdefault(resource_id, []).append(current_time) return await next(event) ```