Key Files in the Pixelle-Video FastAPI API: A Developer Guide
The Pixelle-Video FastAPI API is organized into 11 core modules spanning api/app.py, api/config.py, api/dependencies.py, 11 router files, Pydantic schemas, a task manager, and underlying service implementations in pixelle_video/services/.
The Pixelle-Video repository exposes its AI-powered video generation capabilities through a clean FastAPI layer. For developers building on or extending this API, understanding the key files in the Pixelle-Video FastAPI structure is essential. This guide maps every critical module, from bootstrap to service layer, with direct source links and practical code examples.
Application Bootstrap: api/app.py
The entry point for the entire Pixelle-Video FastAPI server lives in api/app.py. This file instantiates the FastAPI class, wires middleware, and registers all routers under a common prefix.
# api/app.py
app = FastAPI(
title="Pixelle-Video API",
description="…",
version="0.1.0",
docs_url=api_config.docs_url,
redoc_url=api_config.redoc_url,
openapi_url=api_config.openapi_url,
lifespan=lifespan,
)
Key responsibilities include:
- Lifespan management – The
lifespancontext manager starts the globaltask_managerand handles graceful shutdown ofPixelleVideoCore - CORS middleware – Added conditionally based on
api_config.cors_enabled - Router inclusion – All domain routers attach via
app.include_router()withapi_config.api_prefix
Source: [
api/app.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/app.py)
Central Configuration: api/config.py
Runtime settings are consolidated in a single Pydantic model: APIConfig. The globally instantiated api_config object eliminates scattered environment variable access across the codebase.
Configuration categories include:
| Category | Settings |
|---|---|
| Server | host, port, reload |
| CORS | cors_enabled, cors_origins |
| Task limits | max_concurrent_tasks, task_cleanup_interval, task_retention_time |
| Uploads | max_upload_size |
| API paths | api_prefix, docs_url, redoc_url, openapi_url |
Any module can import settings directly:
from api.config import api_config
timeout = api_config.task_retention_time
Source: [
api/config.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/config.py)
Dependency Injection: api/dependencies.py
The Pixelle-Video FastAPI API uses FastAPI's Depends pattern to provide a singleton PixelleVideoCore instance to every request. This pattern ensures expensive initialization happens once and resources are properly released.
async def get_pixelle_video() -> PixelleVideoCore:
global _pixelle_video_instance
if _pixelle_video_instance is None:
_pixelle_video_instance = PixelleVideoCore()
await _pixelle_video_instance.initialize()
return _pixelle_video_instance
The shutdown_pixelle_video hook handles cleanup, including closing the headless browser used by the HTML frame generator.
Source: [
api/dependencies.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/dependencies.py)
API Routers: The Endpoint Layer
All FastAPI route handlers live in api/routers/. Each file encapsulates a functional domain, following the pattern:
from fastapi import APIRouter
router = APIRouter(prefix="/video", tags=["Video"])
@router.post("/generate/sync")
async def generate_video_sync(...):
...
Critical Router Files
Video Router Deep Dive
The [api/routers/video.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/routers/video.py) file contains the most critical endpoints. The sync endpoint:
@router.post("/generate/sync", response_model=VideoGenerateResponse)
async def generate_video_sync(
request_body: VideoGenerateRequest,
pixelle_video: PixelleVideoDep,
request: Request
):
# Resolve frame template → media size
# Build video_params dict
# Call pixelle_video.generate_video(**video_params)
# Return URL, duration, file size
Pydantic Schemas: api/schemas/
All request and response validation uses Pydantic models in api/schemas/. These drive automatic OpenAPI documentation and runtime type checking.
| Schema file | Key models |
|---|---|
[api/schemas/video.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/schemas/video.py) |
VideoGenerateRequest, VideoGenerateResponse, VideoGenerateAsyncResponse |
[api/schemas/llm.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/schemas/llm.py) |
LLM request/response models |
[api/schemas/tts.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/schemas/tts.py) |
TTS models |
[api/schemas/image.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/schemas/image.py) |
Image generation payloads |
[api/schemas/content.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/schemas/content.py) |
Content pipeline structures |
[api/schemas/base.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/schemas/base.py) |
Shared base fields (pagination, etc.) |
Background Task Management: api/tasks/
The async video endpoint relies on a custom task manager rather than Celery or RQ. The implementation lives in two files:
| File | Purpose |
|---|---|
[api/tasks/manager.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/tasks/manager.py) |
In-memory task queue, concurrency limits, cleanup |
[api/tasks/models.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/tasks/models.py) |
Task data structure and status enums |
The manager handles:
- Concurrency limits via
max_concurrent_tasks - Periodic cleanup of finished tasks (
task_cleanup_interval) - Status polling through
GET /tasks/{task_id}
Service Layer: pixelle_video/services/
The routers delegate actual work to the service layer. These implementations contain the heavy lifting for AI operations:
| Service file | Responsibility |
|---|---|
[pixelle_video/services/video.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/pixelle_video/services/video.py) |
End-to-end video pipeline: frame generation, image creation, TTS, stitching |
[pixelle_video/services/llm_service.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/pixelle_video/services/llm_service.py) |
LLM provider communication |
[pixelle_video/services/tts_service.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/pixelle_video/services/tts_service.py) |
Speech synthesis pipelines |
[pixelle_video/services/frame_html.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/pixelle_video/services/frame_html.py) |
HTML template parsing for media size extraction |
[pixelle_video/services/image_analysis.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/pixelle_video/services/image_analysis.py) |
AI image generation logic |
These services are instantiated once through PixelleVideoCore and reused across requests.
Core Service Entry Point: pixelle_video/service.py
The PixelleVideoCore class in [pixelle_video/service.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/pixelle_video/service.py) aggregates all sub-services and exposes the primary generate_video method. It handles:
- Configuration file loading
- Service initialization
- Resource cleanup on shutdown
Practical Development Examples
Starting the FastAPI Server Locally
uv run python api/app.py --host 0.0.0.0 --port 8000 --reload
The --reload flag enables hot-reloading during development.
Calling the Synchronous Video Endpoint
curl -X POST http://localhost:8000/api/video/generate/sync \
-H "Content-Type: application/json" \
-d '{
"text": "The rise of AI in everyday life.",
"mode": "generate",
"n_scenes": 4,
"frame_template": "1080x1920/default.html"
}'
Sample response:
{
"success": true,
"message": "Success",
"video_url": "http://localhost:8000/api/files/20241012_101523/final.mp4",
"duration": 12.3,
"file_size": 8423152
}
Submitting and Polling an Async Video Job
# Submit async job
task_id=$(curl -s -X POST http://localhost:8000/api/video/generate/async \
-H "Content-Type: application/json" \
-d '{"text":"AI future","mode":"generate","frame_template":"1080x1920/default.html"}' \
| jq -r .task_id)
# Poll until complete
while true; do
status=$(curl -s http://localhost:8000/api/tasks/$task_id | jq -r .status)
echo "Task status: $status"
[[ "$status" == "completed" ]] && break
sleep 5
done
# Retrieve result
curl http://localhost:8000/api/tasks/$task_id
Adding a Custom Router
Create api/routers/ping.py:
from fastapi import APIRouter
router = APIRouter(tags=["Utility"])
@router.get("/ping")
async def ping():
return {"message": "pong"}
Register in api/app.py:
from api.routers.ping import router as ping_router
app.include_router(ping_router, prefix=api_config.api_prefix)
The endpoint becomes available at GET /api/ping.
Summary
api/app.py– FastAPI instance creation, middleware, router registrationapi/config.py– Centralized Pydantic settingsapi/dependencies.py– SingletonPixelleVideoCoreinjection with lifecycle managementapi/routers/*.py– 11 domain routers handling HTTP endpointsapi/schemas/*.py– Pydantic models for validation and documentationapi/tasks/– In-memory background task queue with concurrency controlspixelle_video/services/*.py– Core AI service implementationspixelle_video/service.py– Aggregated service core exposinggenerate_video
Frequently Asked Questions
Where is the FastAPI application created in Pixelle-Video?
The FastAPI application is instantiated in [api/app.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/app.py). This file creates the FastAPI object, configures middleware, sets up the lifespan manager for startup/shutdown events, and includes all routers with the configured API prefix.
How does Pixelle-Video handle configuration across the API?
All configuration is centralized in [api/config.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/config.py) through a single APIConfig Pydantic model. A global api_config instance is imported throughout the codebase for consistent access to server settings, CORS rules, task limits, and API path configurations.
What is the difference between synchronous and asynchronous video generation endpoints?
The synchronous endpoint at POST /video/generate/sync ([api/routers/video.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/routers/video.py)) blocks until video generation completes and returns the final result directly. The asynchronous endpoint at POST /video/generate/async creates a background task (managed by [api/tasks/manager.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/tasks/manager.py)) and returns a task_id for polling via GET /tasks/{task_id}.
How do I add a new endpoint to the Pixelle-Video API?
Create a new file in api/routers/ with an APIRouter instance, define your endpoints, then import and register the router in [api/app.py](https://github.com/AIDC-AI/Pixelle-Video/blob/main/api/app.py) using app.include_router() with the api_config.api_prefix. For request/response validation, add corresponding Pydantic models in api/schemas/.
Have a question about this repo?
These articles cover the highlights, but your codebase questions are specific. Give your agent direct access to the source. Share this with your agent to get started:
curl -s "https://instagit.com/install.md" Maintain an open-source project? Get it listed too →