REST API¶

FastAPI application serving attribution data.

Application Factory¶

app ¶

FastAPI application factory for the Music Attribution API.

Provides the create_app factory function and the lifespan async context manager that boots the async database engine on startup and disposes it on shutdown. All route modules (attribution, permissions, health, metrics, CopilotKit) are registered here.

The application follows the attribution-by-design philosophy described in the companion paper (Teikari 2026, Section 4) — provenance metadata is embedded at creation time rather than retrofitted post-hoc.

Notes

CORS origins are read from Settings.cors_origins (comma-separated). The database engine is stored on app.state so that route-level dependencies can access it without global singletons.

lifespan `async` ¶

lifespan(app: FastAPI) -> AsyncGenerator[None]

Manage the async database engine lifecycle.

Creates the SQLAlchemy async engine and session factory on startup, attaches them to app.state, and disposes the engine on shutdown.

PARAMETER	DESCRIPTION
`app`	The FastAPI application instance whose `state` will be populated with `async_engine`, `async_session_factory`, and `settings`. TYPE: `FastAPI`

YIELDS	DESCRIPTION
`None`	Control is yielded to the application between startup and shutdown.

Source code in src/music_attribution/api/app.py

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
    """Manage the async database engine lifecycle.

    Creates the SQLAlchemy async engine and session factory on startup,
    attaches them to ``app.state``, and disposes the engine on shutdown.

    Parameters
    ----------
    app : FastAPI
        The FastAPI application instance whose ``state`` will be populated
        with ``async_engine``, ``async_session_factory``, and ``settings``.

    Yields
    ------
    None
        Control is yielded to the application between startup and shutdown.
    """
    settings = Settings()  # type: ignore[call-arg]

    engine = create_async_engine_factory(settings.database_url)
    factory = async_session_factory(engine)

    app.state.async_engine = engine
    app.state.async_session_factory = factory
    app.state.settings = settings

    db_host = settings.database_url.split("@")[-1] if "@" in settings.database_url else "unknown"
    logger.info("Database engine created: %s", db_host)

    yield

    await engine.dispose()
    logger.info("Database engine disposed")

create_app ¶

create_app() -> FastAPI

Create and configure the FastAPI application.

Instantiates the FastAPI app with metadata, CORS middleware, and all route modules. The lifespan context manager handles database engine startup and shutdown.

RETURNS	DESCRIPTION
`FastAPI`	Fully configured FastAPI application ready to serve.

Notes

Route prefixes:

/health — health check (no prefix)
/metrics — Prometheus metrics (no prefix)
/api/v1/attributions/ — attribution CRUD
/api/v1/permissions/ — permission checks
/api/v1/copilotkit — AG-UI / CopilotKit SSE endpoint

Examples:

>>> from music_attribution.api.app import create_app
>>> app = create_app()
>>> app.title
'Music Attribution API'

Source code in src/music_attribution/api/app.py

def create_app() -> FastAPI:
    """Create and configure the FastAPI application.

    Instantiates the FastAPI app with metadata, CORS middleware, and all
    route modules.  The ``lifespan`` context manager handles database
    engine startup and shutdown.

    Returns
    -------
    FastAPI
        Fully configured FastAPI application ready to serve.

    Notes
    -----
    Route prefixes:

    * ``/health`` — health check (no prefix)
    * ``/metrics`` — Prometheus metrics (no prefix)
    * ``/api/v1/attributions/`` — attribution CRUD
    * ``/api/v1/permissions/`` — permission checks
    * ``/api/v1/copilotkit`` — AG-UI / CopilotKit SSE endpoint

    Examples
    --------
    >>> from music_attribution.api.app import create_app
    >>> app = create_app()
    >>> app.title
    'Music Attribution API'
    """
    app = FastAPI(
        title="Music Attribution API",
        description="REST API for querying music attribution records",
        version=__version__,
        lifespan=lifespan,
    )

    # CORS for frontend dev server — read from Settings once.
    # The same Settings instance is created in lifespan() and stored on
    # app.state; here we only need cors_origins at app creation time.
    cors_settings = Settings()  # type: ignore[call-arg]
    app.add_middleware(
        CORSMiddleware,
        allow_origins=cors_settings.cors_origins.split(","),
        allow_credentials=True,
        allow_methods=["GET", "POST", "OPTIONS"],
        allow_headers=["Content-Type", "Authorization"],
    )

    app.include_router(health_router)
    app.include_router(metrics_router)
    app.include_router(attribution_router, prefix="/api/v1")
    app.include_router(permissions_router, prefix="/api/v1")
    app.include_router(copilotkit_router, prefix="/api/v1")

    return app

Dependencies¶

dependencies ¶

Shared database session helper for route modules.

Provides get_session which extracts the async_sessionmaker from request.app.state and returns a new AsyncSession. This module is the single source of truth for obtaining a database session in API route handlers — no route module should define its own session helper.

The session factory is created in the lifespan context manager (see music_attribution.api.app) and stored on app.state.

get_session ¶

get_session(request: Request) -> AsyncSession

Get an async session from the application's session factory.

PARAMETER	DESCRIPTION
`request`	FastAPI request object with access to `app.state`. TYPE: `Request`

RETURNS	DESCRIPTION
`AsyncSession`	A new async database session. Caller must use it as a context manager (`async with`) to ensure proper cleanup.

Source code in src/music_attribution/api/dependencies.py

def get_session(request: Request) -> AsyncSession:
    """Get an async session from the application's session factory.

    Parameters
    ----------
    request : Request
        FastAPI request object with access to ``app.state``.

    Returns
    -------
    AsyncSession
        A new async database session. Caller must use it as a context
        manager (``async with``) to ensure proper cleanup.
    """
    factory: async_sessionmaker[AsyncSession] = request.app.state.async_session_factory
    return factory()

Attribution Routes¶

attribution ¶

Attribution query endpoints backed by async PostgreSQL repository.

Provides CRUD-style read endpoints for attribution records:

GET /api/v1/attributions/work/{work_id} — single attribution by work ID
GET /api/v1/attributions/ — paginated list with optional filters
GET /api/v1/attributions/{attribution_id}/provenance — provenance chain
GET /api/v1/attributions/search — hybrid search via RRF fusion

All endpoints return JSON representations of AttributionRecord domain objects. The provenance endpoint exposes the full evidence chain with uncertainty metadata, enabling Perplexity-style inline source references (see companion paper, Section 5.2).

Notes

Sessions are obtained from app.state.async_session_factory via the shared get_session helper in dependencies.

get_attribution_by_work_id `async` ¶

get_attribution_by_work_id(
    request: Request, work_id: UUID
) -> dict

Get a single attribution record by work entity ID.

GET /api/v1/attributions/work/{work_id}

Looks up the attribution record associated with a specific musical work entity. Returns 404 if no attribution exists for the given ID.

PARAMETER	DESCRIPTION
`request`	FastAPI request with access to `app.state`. TYPE: `Request`
`work_id`	UUID of the work entity to look up. TYPE: `UUID`

RETURNS	DESCRIPTION
`dict`	JSON-serializable attribution record.

RAISES	DESCRIPTION
`HTTPException`	404 if no attribution is found for the given `work_id`.

Source code in src/music_attribution/api/routes/attribution.py

@router.get("/attributions/work/{work_id}")
async def get_attribution_by_work_id(
    request: Request,
    work_id: uuid.UUID,
) -> dict:
    """Get a single attribution record by work entity ID.

    ``GET /api/v1/attributions/work/{work_id}``

    Looks up the attribution record associated with a specific musical
    work entity.  Returns 404 if no attribution exists for the given ID.

    Parameters
    ----------
    request : Request
        FastAPI request with access to ``app.state``.
    work_id : uuid.UUID
        UUID of the work entity to look up.

    Returns
    -------
    dict
        JSON-serializable attribution record.

    Raises
    ------
    HTTPException
        404 if no attribution is found for the given ``work_id``.
    """
    repo = AsyncAttributionRepository()
    async with get_session(request) as session:
        record = await repo.find_by_work_entity_id(work_id, session)
        if record is None:
            raise HTTPException(status_code=404, detail="Attribution not found")
        return record.model_dump(mode="json")

list_attributions `async` ¶

list_attributions(
    request: Request,
    limit: int = Query(default=50, ge=1, le=100),
    offset: int = Query(default=0, ge=0),
    needs_review: bool | None = Query(default=None),
    assurance_level: str | None = Query(default=None),
) -> list[dict]

List attribution records with pagination, filtering, and sorting.

GET /api/v1/attributions/

Returns attribution records ordered by confidence score (descending). Supports pagination via limit/offset and optional filtering by needs_review flag or assurance_level tier (A0-A3).

PARAMETER	DESCRIPTION
`request`	FastAPI request with access to `app.state`. TYPE: `Request`
`limit`	Maximum number of records to return (1-100), by default 50. TYPE: `int` DEFAULT: `Query(default=50, ge=1, le=100)`
`offset`	Number of records to skip for pagination, by default 0. TYPE: `int` DEFAULT: `Query(default=0, ge=0)`
`needs_review`	If `True`, return only records flagged for human review. If `None` (default), no review filter is applied. TYPE: `bool or None` DEFAULT: `Query(default=None)`
`assurance_level`	Filter by assurance level value (e.g., `"LEVEL_3"` for artist-verified). Maps to the A0-A3 tiered provenance system described in the companion paper (Section 3). TYPE: `str or None` DEFAULT: `Query(default=None)`

RETURNS	DESCRIPTION
`list[dict]`	JSON-serializable list of attribution records sorted by confidence score descending.

Source code in src/music_attribution/api/routes/attribution.py

@router.get("/attributions/")
async def list_attributions(
    request: Request,
    limit: int = Query(default=50, ge=1, le=100),
    offset: int = Query(default=0, ge=0),
    needs_review: bool | None = Query(default=None),
    assurance_level: str | None = Query(default=None),
) -> list[dict]:
    """List attribution records with pagination, filtering, and sorting.

    ``GET /api/v1/attributions/``

    Returns attribution records ordered by confidence score (descending).
    Supports pagination via ``limit``/``offset`` and optional filtering
    by ``needs_review`` flag or ``assurance_level`` tier (A0-A3).

    Parameters
    ----------
    request : Request
        FastAPI request with access to ``app.state``.
    limit : int, optional
        Maximum number of records to return (1-100), by default 50.
    offset : int, optional
        Number of records to skip for pagination, by default 0.
    needs_review : bool or None, optional
        If ``True``, return only records flagged for human review.
        If ``None`` (default), no review filter is applied.
    assurance_level : str or None, optional
        Filter by assurance level value (e.g., ``"LEVEL_3"`` for
        artist-verified).  Maps to the A0-A3 tiered provenance system
        described in the companion paper (Section 3).

    Returns
    -------
    list[dict]
        JSON-serializable list of attribution records sorted by
        confidence score descending.
    """
    repo = AsyncAttributionRepository()
    async with get_session(request) as session:
        if needs_review is True:
            records = await repo.find_needs_review(limit=limit, session=session)
            return [r.model_dump(mode="json") for r in records]

        all_records = await repo.list_all(offset=offset, limit=limit, session=session)

        # Apply assurance_level filter if specified
        if assurance_level is not None:
            all_records = [r for r in all_records if r.assurance_level.value == assurance_level]

        # Sort by confidence descending
        all_records.sort(key=lambda r: r.confidence_score, reverse=True)

        return [r.model_dump(mode="json") for r in all_records]

get_provenance `async` ¶

get_provenance(
    request: Request, attribution_id: UUID
) -> dict

Get the full provenance chain with uncertainty metadata.

GET /api/v1/attributions/{attribution_id}/provenance

Returns the ordered provenance chain (evidence trail) and the uncertainty summary for a specific attribution record. This data enables Perplexity-style inline source references in the frontend, where each claim can be traced back to its originating data source.

The uncertainty summary includes conformal prediction intervals and source agreement metrics (see companion paper, Section 5.2).

PARAMETER	DESCRIPTION
`request`	FastAPI request with access to `app.state`. TYPE: `Request`
`attribution_id`	UUID of the attribution record whose provenance is requested. TYPE: `UUID`

RETURNS	DESCRIPTION
`dict`	Dictionary with keys: `attribution_id` : str — UUID as string. `provenance_chain` : list[dict] — ordered evidence entries. `uncertainty_summary` : dict or None — calibration metadata.

RAISES	DESCRIPTION
`HTTPException`	404 if no attribution record exists for the given ID.

Source code in src/music_attribution/api/routes/attribution.py

@router.get("/attributions/{attribution_id}/provenance")
async def get_provenance(
    request: Request,
    attribution_id: uuid.UUID,
) -> dict:
    """Get the full provenance chain with uncertainty metadata.

    ``GET /api/v1/attributions/{attribution_id}/provenance``

    Returns the ordered provenance chain (evidence trail) and the
    uncertainty summary for a specific attribution record.  This data
    enables *Perplexity-style* inline source references in the frontend,
    where each claim can be traced back to its originating data source.

    The uncertainty summary includes conformal prediction intervals and
    source agreement metrics (see companion paper, Section 5.2).

    Parameters
    ----------
    request : Request
        FastAPI request with access to ``app.state``.
    attribution_id : uuid.UUID
        UUID of the attribution record whose provenance is requested.

    Returns
    -------
    dict
        Dictionary with keys:

        * ``attribution_id`` : str — UUID as string.
        * ``provenance_chain`` : list[dict] — ordered evidence entries.
        * ``uncertainty_summary`` : dict or None — calibration metadata.

    Raises
    ------
    HTTPException
        404 if no attribution record exists for the given ID.
    """
    repo = AsyncAttributionRepository()
    async with get_session(request) as session:
        record = await repo.find_by_id(attribution_id, session)
        if record is None:
            raise HTTPException(status_code=404, detail="Attribution not found")
        return {
            "attribution_id": str(record.attribution_id),
            "provenance_chain": [e.model_dump(mode="json") for e in record.provenance_chain],
            "uncertainty_summary": (
                record.uncertainty_summary.model_dump(mode="json") if record.uncertainty_summary else None
            ),
        }

search_attributions `async` ¶

search_attributions(
    request: Request,
    q: str = Query(
        default="",
        max_length=500,
        description="Search query",
    ),
    limit: int = Query(default=20, ge=1, le=100),
) -> list[dict]

Hybrid search across attribution records using RRF fusion.

GET /api/v1/attributions/search

Combines three retrieval signals via Reciprocal Rank Fusion (RRF):

Full-text search (PostgreSQL tsvector)
Vector similarity (pgvector cosine distance)
Graph context (entity relationship traversal)

This multi-signal approach reduces the risk of any single retrieval method dominating results.

PARAMETER	DESCRIPTION
`request`	FastAPI request with access to `app.state`. TYPE: `Request`
`q`	Search query string, by default `""`. TYPE: `str` DEFAULT: `Query(default='', max_length=500, description='Search query')`
`limit`	Maximum number of results to return (1-100), by default 20. TYPE: `int` DEFAULT: `Query(default=20, ge=1, le=100)`

RETURNS	DESCRIPTION
`list[dict]`	List of dictionaries, each containing: `attribution` : dict — full attribution record. `rrf_score` : float — fused relevance score (higher is better).

Source code in src/music_attribution/api/routes/attribution.py

@router.get("/attributions/search")
async def search_attributions(
    request: Request,
    q: str = Query(default="", max_length=500, description="Search query"),
    limit: int = Query(default=20, ge=1, le=100),
) -> list[dict]:
    """Hybrid search across attribution records using RRF fusion.

    ``GET /api/v1/attributions/search``

    Combines three retrieval signals via Reciprocal Rank Fusion (RRF):

    1. Full-text search (PostgreSQL ``tsvector``)
    2. Vector similarity (pgvector cosine distance)
    3. Graph context (entity relationship traversal)

    This multi-signal approach reduces the risk of any single retrieval
    method dominating results.

    Parameters
    ----------
    request : Request
        FastAPI request with access to ``app.state``.
    q : str, optional
        Search query string, by default ``""``.
    limit : int, optional
        Maximum number of results to return (1-100), by default 20.

    Returns
    -------
    list[dict]
        List of dictionaries, each containing:

        * ``attribution`` : dict — full attribution record.
        * ``rrf_score`` : float — fused relevance score (higher is better).
    """
    from music_attribution.search.hybrid_search import HybridSearchService

    service = HybridSearchService()
    async with get_session(request) as session:
        results = await service.search(q, limit=limit, session=session)
        return [
            {
                "attribution": r.record.model_dump(mode="json"),
                "rrf_score": r.rrf_score,
            }
            for r in results
        ]

Permission Routes¶

permissions ¶

Permission check endpoints backed by async PostgreSQL repository.

Provides MCP-compatible permission query endpoints:

POST /api/v1/permissions/check — check a single permission
GET /api/v1/permissions/{entity_id} — list all permission bundles

These endpoints implement machine-readable permission queries for AI training rights, following the MCP as consent infrastructure model described in the companion paper (Section 6). Each permission check is audit-logged for transparency and compliance.

Notes

Permission bundles use the PermissionTypeEnum to represent training, derivative-work, commercial-use, and other permission types. The audit trail records who asked, when, and with what context.

PermissionCheckRequest ¶

Bases: BaseModel

Request body for a permission check query.

ATTRIBUTE	DESCRIPTION
`entity_id`	UUID of the entity whose permissions are being queried (e.g., a recording or work entity). TYPE: `UUID`
`permission_type`	Permission type string matching a `PermissionTypeEnum` value (e.g., `"AI_TRAINING"`, `"DERIVATIVE_WORK"`). TYPE: `str`
`scope_entity_id`	Optional UUID to scope the permission check to a specific context (e.g., a particular AI model or service). TYPE: `UUID or None`
`requester_id`	Identifier for the requesting party, by default `"anonymous"`. Used for audit logging. TYPE: `str`

PermissionCheckResponse ¶

Bases: BaseModel

Response body for a permission check query.

ATTRIBUTE	DESCRIPTION
`entity_id`	UUID of the entity whose permission was checked. TYPE: `UUID`
`permission_type`	The permission type that was queried. TYPE: `str`
`result`	Permission result value (e.g., `"ALLOW"`, `"DENY"`, `"UNKNOWN"`). TYPE: `str`

check_permission `async` ¶

check_permission(
    request: Request, body: PermissionCheckRequest
) -> PermissionCheckResponse

Check a specific permission for an entity.

POST /api/v1/permissions/check

Queries the permission repository for the given entity and permission type, then records an audit log entry. This implements the machine-readable consent query pattern described in the companion paper (Section 6).

PARAMETER	DESCRIPTION
`request`	FastAPI request with access to `app.state`. TYPE: `Request`
`body`	JSON request body containing the entity ID, permission type, optional scope entity, and requester identifier. TYPE: `PermissionCheckRequest`

RETURNS	DESCRIPTION
`PermissionCheckResponse`	The permission check result with entity ID, type, and outcome.

Source code in src/music_attribution/api/routes/permissions.py

@router.post("/permissions/check")
async def check_permission(
    request: Request,
    body: PermissionCheckRequest,
) -> PermissionCheckResponse:
    """Check a specific permission for an entity.

    ``POST /api/v1/permissions/check``

    Queries the permission repository for the given entity and
    permission type, then records an audit log entry.  This implements
    the machine-readable consent query pattern described in the
    companion paper (Section 6).

    Parameters
    ----------
    request : Request
        FastAPI request with access to ``app.state``.
    body : PermissionCheckRequest
        JSON request body containing the entity ID, permission type,
        optional scope entity, and requester identifier.

    Returns
    -------
    PermissionCheckResponse
        The permission check result with entity ID, type, and outcome.
    """
    repo = AsyncPermissionRepository()
    try:
        perm_type = PermissionTypeEnum(body.permission_type)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=f"Invalid permission type: {body.permission_type}") from exc

    async with get_session(request) as session:
        result = await repo.check_permission(
            entity_id=body.entity_id,
            permission_type=perm_type,
            scope_entity_id=body.scope_entity_id,
            session=session,
        )

        # Find the permission bundle for audit logging
        bundles = await repo.find_by_entity_id(body.entity_id, session)
        if bundles:
            await repo.record_audit(
                permission_id=bundles[0].permission_id,
                requester_id=body.requester_id,
                requester_type="api",
                permission_type=perm_type,
                result=result,
                request_context={
                    "source": "api",
                    "scope_entity_id": str(body.scope_entity_id) if body.scope_entity_id else None,
                },
                session=session,
            )
            await session.commit()

    return PermissionCheckResponse(
        entity_id=body.entity_id,
        permission_type=body.permission_type,
        result=result.value,
    )

list_permissions `async` ¶

list_permissions(
    request: Request, entity_id: UUID
) -> list[dict]

List all permission bundles for an entity.

GET /api/v1/permissions/{entity_id}

Returns every permission bundle associated with the given entity UUID. Each bundle contains the full set of permissions (training, derivative work, commercial use, etc.) and their current status.

PARAMETER	DESCRIPTION
`request`	FastAPI request with access to `app.state`. TYPE: `Request`
`entity_id`	UUID of the entity whose permission bundles are requested. TYPE: `UUID`

RETURNS	DESCRIPTION
`list[dict]`	JSON-serializable list of permission bundle dictionaries.

RAISES	DESCRIPTION
`HTTPException`	404 if no permission bundles exist for the given entity.

Source code in src/music_attribution/api/routes/permissions.py

@router.get("/permissions/{entity_id}")
async def list_permissions(
    request: Request,
    entity_id: uuid.UUID,
) -> list[dict]:
    """List all permission bundles for an entity.

    ``GET /api/v1/permissions/{entity_id}``

    Returns every permission bundle associated with the given entity
    UUID.  Each bundle contains the full set of permissions (training,
    derivative work, commercial use, etc.) and their current status.

    Parameters
    ----------
    request : Request
        FastAPI request with access to ``app.state``.
    entity_id : uuid.UUID
        UUID of the entity whose permission bundles are requested.

    Returns
    -------
    list[dict]
        JSON-serializable list of permission bundle dictionaries.

    Raises
    ------
    HTTPException
        404 if no permission bundles exist for the given entity.
    """
    repo = AsyncPermissionRepository()

    async with get_session(request) as session:
        bundles = await repo.find_by_entity_id(entity_id, session)

    if not bundles:
        raise HTTPException(status_code=404, detail="No permissions found for entity")

    return [b.model_dump(mode="json") for b in bundles]

Health Routes¶

health ¶

Health check endpoint for the Music Attribution API.

Provides a minimal GET /health endpoint used by container orchestrators (Docker health checks, Kubernetes liveness probes) and uptime monitors to verify the service is running.

The endpoint does not check database connectivity — it only confirms that the FastAPI process is alive and able to serve HTTP responses. For deeper health checks, see the /metrics endpoint.

health_check `async` ¶

health_check() -> dict[str, str]

Return a simple health status for the service.

GET /health

Returns a static JSON object confirming the service is alive. This endpoint is intentionally cheap — no database queries, no external calls.

RETURNS	DESCRIPTION
`dict[str, str]`	Dictionary with `status` (always `"healthy"`) and `service` (always `"music-attribution-api"`).

Examples:

>>> import httpx
>>> resp = httpx.get("http://localhost:8000/health")
>>> resp.json()
{'status': 'healthy', 'service': 'music-attribution-api'}

Source code in src/music_attribution/api/routes/health.py

@router.get("/health")
async def health_check() -> dict[str, str]:
    """Return a simple health status for the service.

    ``GET /health``

    Returns a static JSON object confirming the service is alive.
    This endpoint is intentionally cheap — no database queries, no
    external calls.

    Returns
    -------
    dict[str, str]
        Dictionary with ``status`` (always ``"healthy"``) and
        ``service`` (always ``"music-attribution-api"``).

    Examples
    --------
    >>> import httpx
    >>> resp = httpx.get("http://localhost:8000/health")
    >>> resp.json()
    {'status': 'healthy', 'service': 'music-attribution-api'}
    """
    return {"status": "healthy", "service": "music-attribution-api"}

Metrics Routes¶

metrics ¶

Prometheus metrics endpoint for the Music Attribution API.

Exposes GET /metrics in Prometheus exposition format for scraping by Prometheus, Grafana Agent, or any OpenMetrics-compatible collector.

The endpoint is excluded from the OpenAPI schema (include_in_schema=False) so it does not appear in the Swagger UI or auto-generated client stubs.

Notes

Metrics are collected via the prometheus_client default registry (REGISTRY). Custom counters and histograms for attribution pipeline stages can be registered elsewhere and will automatically appear here.

metrics `async` ¶

metrics() -> Response

Serve Prometheus metrics in exposition format.

GET /metrics

Serialises all registered Prometheus collectors into the plain-text exposition format (version 0.0.4). The response content type is set to text/plain; version=0.0.4; charset=utf-8 as required by the Prometheus scraping protocol.

RETURNS	DESCRIPTION
`Response`	FastAPI `Response` with Prometheus-formatted metrics body and the correct content type header.

Notes

This endpoint is hidden from the OpenAPI schema to avoid cluttering the public API documentation.

Source code in src/music_attribution/api/routes/metrics.py

@router.get("/metrics", include_in_schema=False)
async def metrics() -> Response:
    """Serve Prometheus metrics in exposition format.

    ``GET /metrics``

    Serialises all registered Prometheus collectors into the plain-text
    exposition format (version 0.0.4).  The response content type is
    set to ``text/plain; version=0.0.4; charset=utf-8`` as required by
    the Prometheus scraping protocol.

    Returns
    -------
    Response
        FastAPI ``Response`` with Prometheus-formatted metrics body and
        the correct content type header.

    Notes
    -----
    This endpoint is hidden from the OpenAPI schema to avoid cluttering
    the public API documentation.
    """
    return Response(
        content=generate_latest(REGISTRY),
        media_type="text/plain; version=0.0.4; charset=utf-8",
    )

REST API¶

Application Factory¶

app ¶

lifespan async ¶

create_app ¶

Dependencies¶

dependencies ¶

get_session ¶

Attribution Routes¶

attribution ¶

get_attribution_by_work_id async ¶

list_attributions async ¶

get_provenance async ¶

search_attributions async ¶

Permission Routes¶

permissions ¶

PermissionCheckRequest ¶

PermissionCheckResponse ¶

check_permission async ¶

list_permissions async ¶

Health Routes¶

health ¶

health_check async ¶

Metrics Routes¶

metrics ¶

metrics async ¶

lifespan `async` ¶

get_attribution_by_work_id `async` ¶

list_attributions `async` ¶

get_provenance `async` ¶

search_attributions `async` ¶

check_permission `async` ¶

list_permissions `async` ¶

health_check `async` ¶

metrics `async` ¶