api/lyric_search/models.py

"""
Database models for LRCLib lyrics cache.
"""

import os
import urllib.parse
from typing import Type, AsyncGenerator
from sqlalchemy import (
    Column,
    Integer,
    String,
    Float,
    Boolean,
    DateTime,
    ForeignKey,
    UniqueConstraint,
)
from sqlalchemy.orm import relationship, foreign
from sqlalchemy.ext.declarative import declarative_base, DeclarativeMeta
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine, AsyncSession
from sqlalchemy.ext.asyncio import async_sessionmaker

Base: Type[DeclarativeMeta] = declarative_base()


class Tracks(Base):  # type: ignore
    """Tracks table - stores track metadata."""

    __tablename__ = "tracks"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, index=True)
    name_lower = Column(String, index=True)
    artist_name = Column(String, index=True)
    artist_name_lower = Column(String, index=True)
    album_name = Column(String)
    album_name_lower = Column(String, index=True)
    duration = Column(Float, index=True)
    last_lyrics_id = Column(Integer, ForeignKey("lyrics.id"), index=True)
    created_at = Column(DateTime)
    updated_at = Column(DateTime)

    # Relationships
    lyrics = relationship(
        "Lyrics",
        back_populates="track",
        foreign_keys=[last_lyrics_id],
        primaryjoin="Tracks.id == foreign(Lyrics.track_id)",
    )

    # Constraints
    __table_args__ = (
        UniqueConstraint(
            "name_lower",
            "artist_name_lower",
            "album_name_lower",
            "duration",
            name="uq_tracks",
        ),
    )


class Lyrics(Base):  # type: ignore
    """Lyrics table - stores lyrics content."""

    __tablename__ = "lyrics"

    id = Column(Integer, primary_key=True, autoincrement=True)
    plain_lyrics = Column(String)
    synced_lyrics = Column(String)
    track_id = Column(Integer, ForeignKey("tracks.id"), index=True)
    has_plain_lyrics = Column(Boolean, index=True)
    has_synced_lyrics = Column(Boolean, index=True)
    instrumental = Column(Boolean)
    source = Column(String, index=True)
    created_at = Column(DateTime, index=True)
    updated_at = Column(DateTime)

    # Relationships
    track = relationship(
        "Tracks",
        back_populates="lyrics",
        foreign_keys=[track_id],
        primaryjoin=(Tracks.id == foreign(track_id)),
        remote_side=Tracks.id,
    )


# PostgreSQL connection - using environment variables
POSTGRES_HOST = os.getenv("POSTGRES_HOST", "localhost")
POSTGRES_PORT = os.getenv("POSTGRES_PORT", "5432")
POSTGRES_DB = os.getenv("POSTGRES_DB", "lrclib")
POSTGRES_USER = os.getenv("POSTGRES_USER", "api")
POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD", "")

# URL-encode the password to handle special characters
encoded_password = urllib.parse.quote_plus(POSTGRES_PASSWORD)

DATABASE_URL: str = f"postgresql+asyncpg://{POSTGRES_USER}:{encoded_password}@{POSTGRES_HOST}:{POSTGRES_PORT}/{POSTGRES_DB}"
async_engine: AsyncEngine = create_async_engine(
    DATABASE_URL, pool_size=20, max_overflow=10, pool_pre_ping=True, echo=False
)
AsyncSessionLocal = async_sessionmaker(bind=async_engine, expire_on_commit=False)


async def get_async_db():
    """Get async database session."""
    async with AsyncSessionLocal() as session:
        yield session