Aniworld/src/server/database

Database Layer

SQLAlchemy-based database layer for the Aniworld web application.

Overview

This package provides persistent storage for anime series, episodes, download queue, and user sessions using SQLAlchemy ORM with comprehensive service layer for CRUD operations.

Quick Start

Installation

Install required dependencies:

pip install sqlalchemy aiosqlite

Or use the project requirements:

pip install -r requirements.txt

Initialization

Initialize the database on application startup:

from src.server.database import init_db, close_db

# Startup
await init_db()

# Shutdown
await close_db()

Usage in FastAPI

Use the database session dependency in your endpoints:

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from src.server.database import get_db_session, AnimeSeries
from sqlalchemy import select

@app.get("/anime")
async def get_anime(db: AsyncSession = Depends(get_db_session)):
    result = await db.execute(select(AnimeSeries))
    return result.scalars().all()

Models

AnimeSeries

Represents an anime series with metadata and relationships.

series = AnimeSeries(
    key="attack-on-titan",
    name="Attack on Titan",
    site="https://aniworld.to",
    folder="/anime/attack-on-titan",
    description="Epic anime about titans",
    status="completed",
    total_episodes=75
)

Episode

Individual episodes linked to series.

episode = Episode(
    series_id=series.id,
    season=1,
    episode_number=5,
    title="The Fifth Episode",
    is_downloaded=True
)

DownloadQueueItem

Download queue with progress tracking.

from src.server.database.models import DownloadStatus, DownloadPriority

item = DownloadQueueItem(
    series_id=series.id,
    season=1,
    episode_number=3,
    status=DownloadStatus.DOWNLOADING,
    priority=DownloadPriority.HIGH,
    progress_percent=45.5
)

UserSession

User authentication sessions.

from datetime import datetime, timedelta

session = UserSession(
    session_id="unique-session-id",
    token_hash="hashed-jwt-token",
    expires_at=datetime.utcnow() + timedelta(hours=24),
    is_active=True
)

Mixins

TimestampMixin

Adds automatic timestamp tracking:

from src.server.database.base import Base, TimestampMixin

class MyModel(Base, TimestampMixin):
    __tablename__ = "my_table"
    # created_at and updated_at automatically added

SoftDeleteMixin

Provides soft delete functionality:

from src.server.database.base import Base, SoftDeleteMixin

class MyModel(Base, SoftDeleteMixin):
    __tablename__ = "my_table"

    # Usage
    instance.soft_delete()  # Mark as deleted
    instance.is_deleted     # Check if deleted
    instance.restore()      # Restore deleted record

Configuration

Configure database via environment variables:

DATABASE_URL=sqlite:///./data/aniworld.db
LOG_LEVEL=DEBUG  # Enables SQL query logging

Or in code:

from src.config.settings import settings

settings.database_url = "sqlite:///./data/aniworld.db"

Testing

Run database tests:

pytest tests/unit/test_database_models.py -v

The test suite uses an in-memory SQLite database for isolation and speed.

Architecture

• base.py: Base declarative class and mixins
• models.py: SQLAlchemy ORM models (4 models)
• connection.py: Engine, session factory, dependency injection
• **init.py**: Package exports
• service.py: Service layer with CRUD operations

Service Layer

The service layer provides high-level CRUD operations for all models:

AnimeSeriesService

from src.server.database import AnimeSeriesService

# Create series
series = await AnimeSeriesService.create(
    db,
    key="my-anime",
    name="My Anime",
    site="https://example.com",
    folder="/path/to/anime"
)

# Get by ID or key
series = await AnimeSeriesService.get_by_id(db, series_id)
series = await AnimeSeriesService.get_by_key(db, "my-anime")

# Get all with pagination
all_series = await AnimeSeriesService.get_all(db, limit=50, offset=0)

# Update
updated = await AnimeSeriesService.update(db, series_id, name="Updated Name")

# Delete (cascades to episodes and downloads)
deleted = await AnimeSeriesService.delete(db, series_id)

# Search
results = await AnimeSeriesService.search(db, "naruto", limit=10)

EpisodeService

from src.server.database import EpisodeService

# Create episode
episode = await EpisodeService.create(
    db,
    series_id=1,
    season=1,
    episode_number=5,
    title="Episode 5"
)

# Get episodes for series
episodes = await EpisodeService.get_by_series(db, series_id, season=1)

# Get specific episode
episode = await EpisodeService.get_by_episode(db, series_id, season=1, episode_number=5)

# Mark as downloaded
updated = await EpisodeService.mark_downloaded(
    db,
    episode_id,
    file_path="/path/to/file.mp4",
    file_size=1024000
)

DownloadQueueService

from src.server.database import DownloadQueueService
from src.server.database.models import DownloadPriority, DownloadStatus

# Add to queue
item = await DownloadQueueService.create(
    db,
    series_id=1,
    season=1,
    episode_number=5,
    priority=DownloadPriority.HIGH
)

# Get pending downloads (ordered by priority)
pending = await DownloadQueueService.get_pending(db, limit=10)

# Get active downloads
active = await DownloadQueueService.get_active(db)

# Update status
updated = await DownloadQueueService.update_status(
    db,
    item_id,
    DownloadStatus.DOWNLOADING
)

# Update progress
updated = await DownloadQueueService.update_progress(
    db,
    item_id,
    progress_percent=50.0,
    downloaded_bytes=500000,
    total_bytes=1000000,
    download_speed=50000.0
)

# Clear completed
count = await DownloadQueueService.clear_completed(db)

# Retry failed downloads
retried = await DownloadQueueService.retry_failed(db, max_retries=3)

UserSessionService

from src.server.database import UserSessionService
from datetime import datetime, timedelta

# Create session
expires_at = datetime.utcnow() + timedelta(hours=24)
session = await UserSessionService.create(
    db,
    session_id="unique-session-id",
    token_hash="hashed-jwt-token",
    expires_at=expires_at,
    user_id="user123",
    ip_address="127.0.0.1"
)

# Get session
session = await UserSessionService.get_by_session_id(db, "session-id")

# Get active sessions
active = await UserSessionService.get_active_sessions(db, user_id="user123")

# Update activity
updated = await UserSessionService.update_activity(db, "session-id")

# Revoke session
revoked = await UserSessionService.revoke(db, "session-id")

# Cleanup expired sessions
count = await UserSessionService.cleanup_expired(db)

Database Schema

anime_series (id, key, name, site, folder, ...)
├── episodes (id, series_id, season, episode_number, ...)
└── download_queue (id, series_id, season, episode_number, status, ...)

user_sessions (id, session_id, token_hash, expires_at, ...)

Production Considerations

SQLite (Current)

• Single file: data/aniworld.db
• WAL mode for concurrency
• Foreign keys enabled
• Static connection pool

PostgreSQL/MySQL (Future)

For multi-process deployments:

DATABASE_URL=postgresql+asyncpg://user:pass@host/db
# or
DATABASE_URL=mysql+aiomysql://user:pass@host/db

Configure connection pooling:

engine = create_async_engine(
    url,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True
)

Performance Tips

1. Indexes: Models have indexes on frequently queried columns
2. Relationships: Use selectinload() or joinedload() for eager loading
3. Batching: Use bulk operations for multiple inserts/updates
4. Query Optimization: Profile slow queries in DEBUG mode

Example with eager loading:

from sqlalchemy.orm import selectinload

result = await db.execute(
    select(AnimeSeries)
    .options(selectinload(AnimeSeries.episodes))
    .where(AnimeSeries.key == "attack-on-titan")
)
series = result.scalar_one()
# episodes already loaded, no additional queries

Troubleshooting

Database not initialized

RuntimeError: Database not initialized. Call init_db() first.

Solution: Call await init_db() during application startup.

Table does not exist

sqlalchemy.exc.OperationalError: no such table: anime_series

Solution: Base.metadata.create_all() is called automatically by init_db().

Foreign key constraint failed

sqlalchemy.exc.IntegrityError: FOREIGN KEY constraint failed

Solution: Ensure referenced records exist before creating relationships.

Further Reading

• SQLAlchemy 2.0 Documentation
• FastAPI with Databases