Beyond Basic CRUD: Building a Scalable Music API with FastAPI & Clean Architecture

Aslı KARABULUT — Mon, 24 Nov 2025 11:15:38 +0000

🚀 Introduction

When I started building TAM Music, a platform for managing comprehensive music libraries, I knew main.py wouldn't be enough. Real-world applications need structure. They need to handle complex relationships—like Artists having Albums, and Albums having Songs—without crashing into circular dependency errors.

In this post, I’ll walk you through how I refactored a monolithic script into a robust, Service-Oriented REST API using FastAPI. We will dive deep into the architecture, specifically focusing on Dependency Injection, Pydantic v2, and handling Asynchronous I/O with strict validation.

🛠️ The Tech Stack

I chose a modern Python stack to ensure performance, type safety, and cryptographic agility.

FastAPI: Chosen for its high performance (Starlette-based) and native support for asynchronous programming.
Pydantic V2: Used for strict schema validation. We leverage advanced features like model_rebuild() to resolve forward references in the schema graph.
Dependency Injection (DI): FastAPI's native DI system implements the Inversion of Control (IoC) principle, managing our service lifecycles.
Pandas & OpenPyXL: Used for strict content validation of uploaded Excel files.
Aiofiles: For non-blocking asynchronous file writes to the disk.
Passlib (Argon2) & Python-Jose: State-of-the-art security using generic hashing and stateless JWT authentication.

🏗️ Architecture: The Service-Repository Pattern

To keep the code maintainable, I strictly separated concerns. The API is not just a collection of endpoints; it's a layered system.

Folder Structure
Here is the actual structure. Notice how db_store.py acts as our persistence layer, decoupled from the router logic.

TAM_Music/
├── models/             # Pydantic Models (Data Layer)
│   ├── instrument_model.py
│   ├── artist_model.py
│   └── ...
├── routers/            # API Controllers (Presentation Layer)
│   ├── instrument_router.py
│   ├── db_store.py     # In-Memory Database (O(1) access)
│   └── ...
├── services/           # Business Logic (Service Layer)
│   ├── instrument_service.py
│   └── ...
└── main.py             # Application Entry Point

1. The Data Layer: Defining the Models

Before we can store anything, we need to define our data structure. I use Pydantic inheritance to keep things clean. Notice the InstrumentFilter model—I use this specifically for handling search query parameters cleanly!

# models/instrument_model.py
from typing import List, Optional
from pydantic import BaseModel, Field

class InstrumentBase(BaseModel):
    neck_type: str = Field(..., min_length=1) 
    top_back_body: str = Field(..., min_length=1)
    # ... other fields

class InstrumentCreate(InstrumentBase):
    pass

class InstrumentModel(InstrumentBase):
    instrument_id: int = Field(..., description="Instrument PK")
    images: List[str] = Field(default_factory=list, description="Images' Url or Paths")

    class Config:
        from_attributes = True

class InstrumentFilter(BaseModel):
    neck_type: Optional[str] = None
    number_of_frets_min: Optional[int] = Field(None, ge=1)
    number_of_frets_max: Optional[int] = Field(None, le=50)
    # ... filters used for query params

2. The Persistence Layer

Before diving into logic, we need a place to store this data. For this project, I used an In-Memory Strategy using Python Dictionaries.

Why?

Speed: $O(1)$ Time Complexity for lookups, inserts, and deletions.
Focus: It allowed me to focus on the Architecture (Services/Routers) without setting up a heavy DB initially.

# routers/db_store.py
from typing import Dict

# Simulating database tables with Dicts
instruments_db: Dict[int, 'InstrumentModel'] = {}
songs_db: Dict[int, 'SongModel'] = {}
# ... other dbs

def get_next_id(db: Dict):
    return max(db.keys(), default=0) + 1

🧠 Deep Dive: Solving Technical Challenges

1. Handling Circular Dependencies in Pydantic V2

In a relational domain, objects reference each other. An Artist model needs a list of Albums, but an Album model needs the Artist info. In Python, this causes immediate ImportError or recursion issues.

The Solution: I used Forward References (string-based type hints) and Pydantic's model_rebuild() method. This forces Pydantic to defer the evaluation of the type until the entire module is loaded.

# models/song_model.py
from typing import Optional
from pydantic import BaseModel, Field
# ... imports

class SongModel(SongBase):
    song_id: int = Field(..., description="Song PK")
    # Using string forward reference to avoid circular import at runtime
    genre: Optional['GenreSimplified'] = Field(None, description="Related genre")

    class Config:
        from_attributes = True

# Rebuild the model at the end of the file
from .genre_model import GenreSimplified
SongModel.model_rebuild()

2. Service-to-Service Communication

This is where the architecture shines. Instead of instantiating classes manually inside routers, I inject services into other services. This adheres to the Single Responsibility Principle (SRP).

For example, the ArtistService doesn't query the Album database directly. It asks the AlbumService.

# services/artist_service.py
from services.album_service import AlbumService 

class ArtistService:
    def __init__(
        self, 
        artists_db: Dict[int, ArtistModel] = artists_db, 
        # DIP: Injecting AlbumService instead of direct DB access
        album_service: AlbumService = AlbumService() 
    ):
        self.artists_db = artists_db
        self.album_service = album_service

    def get_artist_with_relations(self, artist_id: int) -> Optional[ArtistModel]:
        artist = self.artists_db.get(artist_id)
        if artist is None: return None

        # Solving relations using the injected service
        albums = self.album_service.get_simplified_albums_by_artist(artist_id)

        # ... logic to merge data ...
        return ArtistModel(**artist_dict)

3. Async I/O vs. Strict Validation Trade-off

For the Instrument module, I needed to validate bulk uploads via Excel. Here, I had to make an engineering trade-off:

Pandas is great for validation but is synchronous (CPU-bound).
Aiofiles is great for saving files asynchronously (I/O-bound).

I combined them to ensure the API remains responsive while ensuring data integrity.

# services/instrument_service.py
import aiofiles
import pandas as pd

class InstrumentService:
    # ...
    async def _process_and_save_images(self, images: List[UploadFile]) -> List[str]:
        image_paths = []
        for file in images:
            # 1. Validation (Synchronous but fast for metadata)
            if file.content_type not in ALLOWED_MIME_TYPES:
                raise HTTPException(status_code=415, detail="Invalid file type")

            # 2. Content Check (CPU Bound - Pandas)
            file_content = await file.read()
            if "spreadsheet" in file.content_type:
                 self._check_excel_content(file_content, file.filename)

            # 3. Saving File (I/O Bound - Asynchronous)
            # This prevents blocking the event loop during disk writes
            saved_path = await self._save_upload_file(file_content, file.filename)
            image_paths.append(saved_path)

        return image_paths

🔐 Security: Stateless Authentication

Security isn't an afterthought. In this architecture, I strictly separated the security logic (hashing, JWT creation) from the HTTP interface.

Since listing the entire codebase would be overwhelming, I will focus on the core logic used for authentication.

1. The Security Engine (UserService)

The UserService handles the raw logic. Below, you can see how Argon2 is used for password verification and how JWTs are generated. I've omitted standard CRUD methods (like get_user_by_id) to keep it focused.

# services/user_service.py
from datetime import datetime, timedelta, timezone
from typing import Optional
from jose import jwt
from passlib.context import CryptContext
from models.user_model import UserLogin, UserModel

# Configuration
pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")
SECRET_KEY = "tam-music-super-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

class UserService:
    def __init__(self, db=None):
        self.db = db or {}

    # --- Core Security Logic ---
    def verify_password(self, plain_password, hashed_password):
        return pwd_context.verify(plain_password, hashed_password)

    def create_access_token(self, data: dict):
        to_encode = data.copy()
        expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        to_encode.update({"exp": expire.timestamp()})

        return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

    def authenticate_user(self, user_data: UserLogin) -> Optional[UserModel]:
        # Flexible login: Check by username first, then email
        user = self.get_user_by_username(user_data.username)
        if not user:
            user = self.get_user_by_email(user_data.username)

        if not user or not self.verify_password(user_data.password, user.hashed_password):
            return None

        return user

    # ... Helper methods like get_user_by_username follow here ...

2. The Interface (UserRouter)

The Router acts purely as an interface. Notice how I inject the UserService into the endpoint. This allows us to keep the router clean and testable.

# routers/user_router.py
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from services.user_service import UserService
from models.user_model import Token, UserLogin

router = APIRouter(prefix="/auth", tags=["auth"])

# Dependency Injection
def get_user_service() -> UserService:
    return UserService()

@router.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: OAuth2PasswordRequestForm = Depends(),
    service: UserService = Depends(get_user_service)
):
    """
    Standard OAuth2 compatible token login.
    """
    try:
        # 1. Map form data to internal model
        login_data = UserLogin(username=form_data.username, password=form_data.password) 

        # 2. Delegate authentication to the service
        user = service.authenticate_user(login_data)

        if not user:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Incorrect username or password",
                headers={"WWW-Authenticate": "Bearer"}
            )

        # 3. Create and return the Token
        access_token = service.create_access_token(data={"sub": user.username})
        return Token(access_token=access_token, token_type="bearer")

    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, 
            detail=str(e)
        )

🔌 Bringing It All Together: main.py

This is the entry point of our application. Because we separated our logic into Routers and Services, main.py remains incredibly clean. It simply wires everything up.

# main.py
from fastapi import FastAPI
from routers import album_router, user_router # ... other routers

app = FastAPI(title="TAM Music API")

# Registering the routers
app.include_router(user_router.router)
app.include_router(album_router.router)
# ...

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

👋 Conclusion

Building TAM Music was a journey in understanding how to structure scalable Python applications. By leveraging FastAPI's Dependency Injection and Pydantic's advanced validation, I created an API that is clean, testable, and ready for growth.

I hope this breakdown helps you in structuring your own FastAPI projects! If you have any questions about the architecture or the specific implementation details, let's discuss them in the comments below.

Happy Coding! 🎧

DEV Community: Aslı KARABULUT