Source code for dakweb.routers.source

# SPDX-License-Identifier:  GPL-2.0-or-later
# © 2026, Ansgar 🙀 <ansgar@debian.org>
# © 2026, Anton Gladky <gladk@debian.org>

from fastapi import APIRouter, Depends
from pydantic import BaseModel
from sqlalchemy import or_, select
from sqlalchemy.orm import Session

from daklib.dbconn import (
    DBSource,
    DSCFile,
    MetadataKey,
    PoolFile,
    SourceMetadata,
    Suite,
)
from dakweb.routers.deps import get_db

router = APIRouter()




class DscInSuite(BaseModel):
    version: str
    component: str
    filename: str
    filesize: int
    sha256sum: str





@router.get("/dsc_in_suite/{suite}/{source}")
def dsc_in_suite(
    suite: str,
    source: str,
    db: Session = Depends(get_db),
) -> list[DscInSuite]:
    """Find all dsc files for a given source package in a given suite."""

    stmt = (
        select(DSCFile)
        .join(PoolFile)
        .join(DBSource)
        .join(Suite, DBSource.suites)
        .where(or_(Suite.suite_name == suite, Suite.codename == suite))
        .where(DBSource.source == source)
        .where(PoolFile.filename.endswith(".dsc"))
    )

    return [
        DscInSuite(
            version=p.source.version,
            component=p.poolfile.component.component_name,
            filename=p.poolfile.filename,
            filesize=p.poolfile.filesize,
            sha256sum=p.poolfile.sha256sum,
        )
        for p in db.scalars(stmt)
    ]





class FileInArchive(BaseModel):
    filename: str
    component: str
    sha256sum: str





@router.get("/file_in_archive/{filepattern:path}")
def file_in_archive(
    filepattern: str,
    db: Session = Depends(get_db),
) -> list[FileInArchive]:
    """Check if a file pattern is known to the archive (SQL LIKE syntax)."""

    stmt = select(PoolFile).where(PoolFile.filename.like(filepattern))

    return [
        FileInArchive(
            filename=p.filename,
            component=p.component.component_name,
            sha256sum=p.sha256sum,
        )
        for p in db.scalars(stmt)
    ]





@router.get("/sha256sum_in_archive/{sha256sum}")
def sha256sum_in_archive(
    sha256sum: str,
    db: Session = Depends(get_db),
) -> list[FileInArchive]:
    """Check if files with matching sha256sums are known to the archive."""

    stmt = select(PoolFile).where(PoolFile.sha256sum == sha256sum)

    return [
        FileInArchive(
            filename=p.filename,
            component=p.component.component_name,
            sha256sum=p.sha256sum,
        )
        for p in db.scalars(stmt)
    ]





class SourcesInSuite(BaseModel):
    source: str
    version: str





@router.get("/sources_in_suite/{suite}")
def sources_in_suite(
    suite: str,
    db: Session = Depends(get_db),
) -> list[SourcesInSuite]:
    """Return all source packages and versions in a given suite."""

    stmt = (
        select(DBSource)
        .join(Suite, DBSource.suites)
        .where(or_(Suite.suite_name == suite, Suite.codename == suite))
    )

    return [
        SourcesInSuite(source=p.source, version=p.version) for p in db.scalars(stmt)
    ]





@router.get("/all_sources")
def all_sources(db: Session = Depends(get_db)) -> list[SourcesInSuite]:
    """Return all source packages and versions known to the archive (includes NEW)."""

    return [
        SourcesInSuite(source=p.source, version=p.version)
        for p in db.scalars(select(DBSource))
    ]





class SourceByMetadata(BaseModel):
    source: str
    metadata_value: str





@router.get("/source/by_metadata/{key}")
def source_by_metadata(
    key: str,
    db: Session = Depends(get_db),
) -> list[SourceByMetadata]:
    """Find all Debian source packages which have the specified metadata set."""

    stmt = (
        select(DBSource.source, SourceMetadata.value)
        .join(SourceMetadata)
        .join(MetadataKey)
        .where(MetadataKey.key == key)
    )

    return [
        SourceByMetadata(source=p.source, metadata_value=p.value)
        for p in db.execute(stmt)
    ]