# Key Source Files in MoneyPrinterTurbo: A Complete Architecture Guide

> Explore the key source files of MoneyPrinterTurbo's FastAPI architecture. Understand the bootstrap routing models services utilities configuration and task managers layers for efficient microservice development.

- Repository: [Harry/MoneyPrinterTurbo](https://github.com/harry0703/MoneyPrinterTurbo)
- Tags: architecture
- Published: 2026-03-23

---

**MoneyPrinterTurbo organizes its FastAPI microservice into seven distinct layers—bootstrap, routing, models, services, utilities, configuration, and task managers—with [`app/asgi.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/asgi.py), [`app/controllers/v1/video.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/controllers/v1/video.py), and [`app/services/task.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/services/task.py) serving as the primary orchestration points.**

MoneyPrinterTurbo is an open-source FastAPI microservice that automates short-form video generation, subtitle rendering, and text-to-speech synthesis. Understanding the key source files in MoneyPrinterTurbo is essential for developers who want to extend the pipeline, debug generation failures, or integrate custom voice providers. The codebase follows a clean separation of concerns, with distinct modules handling HTTP routing, Pydantic validation, state management, and asynchronous task execution.

## 1. Application Bootstrap and Entry Points

The bootstrap layer initializes the ASGI server and constructs the FastAPI application instance.

### main.py

Located at the repository root, [`main.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/main.py) serves as the entry point that launches the Uvicorn server. It imports the application factory from `app.asgi` and starts the server with configuration values loaded from [`config.example.toml`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/config.example.toml).

```python

# main.py (simplified)

if __name__ == "__main__":
    uvicorn.run(app="app.asgi:app", host=config.listen_host,
                port=config.listen_port, reload=config.reload_debug)

```

### app/asgi.py

The [`app/asgi.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/asgi.py) module contains the `get_application()` factory function. This function creates the `FastAPI` instance, registers the root router, attaches custom exception handlers for `HttpException` and `RequestValidationError`, mounts static file directories for generated media, and configures CORS middleware.

```python
instance.include_router(root_api_router)          # Registers all API endpoints

instance.add_exception_handler(HttpException, ...) # Custom error handling

instance.mount("/tasks", StaticFiles(...))        # Serves generated media files

instance.mount("/", StaticFiles(...))             # Serves public UI assets

```

## 2. API Routing and Controllers

The routing layer maps HTTP endpoints to controller functions and handles request validation.

### app/router.py

This file declares the root `APIRouter` and includes version-specific sub-routers. It acts as the central hub that aggregates all endpoint definitions before they are registered in the ASGI application.

### app/controllers/v1/video.py

Located at [`app/controllers/v1/video.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/controllers/v1/video.py), this is the primary controller implementing the public API for video creation, subtitle generation, audio synthesis, media upload/download, and task management. It defines endpoints such as `POST /videos` for creating generation tasks.

```python
@router.post("/videos", response_model=TaskResponse)
def create_video(background_tasks: BackgroundTasks, request: Request,
                 body: TaskVideoRequest):
    return create_task(request, body, stop_at="video")

```

The `create_task()` function generates a UUID, records the request ID, updates the shared state, and delegates to the task manager to run `app.services.task.start` asynchronously.

### app/controllers/base.py

This utility module provides helper functions used by controllers, such as extracting the `request-id` header and handling API key validation.

## 3. Data Models and Validation

Pydantic schemas enforce type safety and generate OpenAPI documentation automatically.

### app/models/schema.py

This central file defines all request and response models, including `TaskVideoRequest`, `TaskResponse`, and `VideoMaterialUploadResponse`. These models serve as the core contract between the API layer and the service layer.

### app/models/exception.py

Defines the custom `HttpException` class used throughout the API for consistent error handling.

### app/models/const.py

Contains global constants such as punctuation characters used in text processing.

## 4. Core Services and Business Logic

The service layer implements the heavy-lifting for video generation, state management, and media processing.

### app/services/state.py

This module provides an in-memory task state store that tracks task status, generated file paths, and pagination metadata. Functions like `update_task()` and `get_task()` manage the lifecycle of generation jobs.

### app/services/task.py

Located at [`app/services/task.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/services/task.py), this is the orchestration engine that coordinates the end-to-end video generation pipeline. It handles text-to-speech synthesis, video clipping, concatenation, and subtitle rendering.

### app/services/video.py

Contains helper functions for video-specific processing, such as selecting background music and measuring text dimensions for subtitle placement.

### app/services/voice.py

Manages retrieval of available voice models from providers like Azure, Gemini, and SiliconFlow.

### app/services/material.py

Reads API keys for external services from the configuration file.

## 5. Utility Functions and Helpers

### app/utils/utils.py

This core utility library provides generic helper functions including JSON response builders, UUID generation, filesystem path helpers (`task_dir`, `song_dir`, `public_dir`), background thread runners, string manipulation, and locale loading.

## 6. Configuration and Logging

### app/config/__init__.py

This module loads configuration from [`config.example.toml`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/config.example.toml) and initializes the Loguru logger with colored console output. It exposes configuration values such as `listen_host`, `listen_port`, and `project_name` to the rest of the application.

## 7. Pluggable Task Managers

The application supports multiple concurrency backends through an abstract manager interface.

### app/controllers/manager/base_manager.py

Defines the abstract base class specifying the interface for `add_task`, `cancel_task`, and other management operations.

### app/controllers/manager/memory_manager.py

Implements a simple in-process task queue using Python `threading`. This is the default manager when Redis is not enabled.

### app/controllers/manager/redis_manager.py

Provides a Redis-backed task queue for distributed deployments. Enabled via the `enable_redis` configuration option.

The router automatically selects the appropriate manager based on `config.app["enable_redis"]`, ensuring the rest of the codebase remains agnostic to the underlying concurrency mechanism.

## 8. Web Interface and Documentation

### webui/Main.py

An optional Streamlit-based web interface located at [`webui/Main.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/webui/Main.py) for manual testing and interaction with the API.

### README.md

The primary documentation file containing project overview, installation steps, and quick-start instructions.

## Practical Code Examples

### Starting the Server

To launch the development server, run the entry point script:

```bash

# Install dependencies

pip install -r requirements.txt

# Run with auto-reload enabled

python main.py

```

The service will be available at `http://127.0.0.1:8000` (or the host/port defined in your TOML config). Interactive API documentation is automatically generated at `/docs`.

### Creating a Video Generation Task

Submit a POST request to the videos endpoint defined in [`app/controllers/v1/video.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/controllers/v1/video.py):

```bash
curl -X POST "http://127.0.0.1:8000/videos" \
     -H "Content-Type: application/json" \
     -d '{
           "script": "Hello, this is a demo video generated by Money Printer Turbo.",
           "bgm_type": "random",
           "voice": "en-US-Standard-A"
         }'

```

The controller's `create_video` function returns a `TaskResponse` containing a `task_id` that you can use to track progress.

### Checking Task Status

Poll the task state managed by [`app/services/state.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/services/state.py):

```bash
curl "http://127.0.0.1:8000/tasks/a3f9c2d4-1b6e-4c7a-9f3e-7d5b6a2c1e0f"

```

When the orchestration in [`app/services/task.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/services/task.py) completes, the response includes download URLs for the final video files stored in the static tasks directory.

## Summary

- **Entry Point**: [`main.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/main.py) launches the Uvicorn server, while [`app/asgi.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/asgi.py) constructs the FastAPI application and registers routers.
- **Routing Layer**: [`app/router.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/router.py) aggregates endpoints, with [`app/controllers/v1/video.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/controllers/v1/video.py) handling the core video generation API.
- **Data Validation**: [`app/models/schema.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/models/schema.py) defines Pydantic models for type-safe request/response handling.
- **Business Logic**: [`app/services/task.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/services/task.py) orchestrates the generation pipeline, supported by [`state.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/state.py), [`video.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/video.py), and [`voice.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/voice.py).
- **Concurrency**: Pluggable task managers in `app/controllers/manager/` support both in-memory and Redis-backed queues.
- **Configuration**: [`app/config/__init__.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/config/__init__.py) loads TOML settings and initializes Loguru logging.

## Frequently Asked Questions

### What is the main entry point to start the MoneyPrinterTurbo server?

The server entry point is [`main.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/main.py) in the repository root. This script imports the ASGI application factory from `app.asgi` and launches Uvicorn with configuration values loaded from [`config.example.toml`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/config.example.toml), including host, port, and debug reload settings.

### How does MoneyPrinterTurbo handle asynchronous video generation tasks?

The system uses a pluggable task manager architecture defined in [`app/controllers/manager/base_manager.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/controllers/manager/base_manager.py). By default, [`app/controllers/manager/memory_manager.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/controllers/manager/memory_manager.py) handles tasks using Python threading, while [`app/controllers/manager/redis_manager.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/controllers/manager/redis_manager.py) provides distributed queue capabilities when `enable_redis` is set to true in the configuration.

### Where are the API request and response models defined?

All Pydantic schemas are centralized in [`app/models/schema.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/models/schema.py). This file defines the core data contracts including `TaskVideoRequest`, `TaskResponse`, and `VideoMaterialUploadResponse`, which enforce type safety and automatically generate OpenAPI documentation at the `/docs` endpoint.

### How can I switch from in-memory task management to Redis?

To enable Redis-backed task queues, modify the `enable_redis` setting in your configuration file loaded by [`app/config/__init__.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/config/__init__.py). When set to true, the router in [`app/router.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/router.py) automatically instantiates `RedisManager` from [`app/controllers/manager/redis_manager.py`](https://github.com/harry0703/MoneyPrinterTurbo/blob/main/app/controllers/manager/redis_manager.py) instead of the default `MemoryManager`, allowing distributed task processing across multiple worker instances.