API reference

safir Package

Safir is the Rubin Observatory’s library for building FastAPI services for the Rubin Science Platform.

Variables

__version__	The version string of Safir (PEP 440 / SemVer compatible).
version_info	The decomposed version, split across ".."

safir.arq Module

An arq client with a mock for testing.

Classes

ArqJobError(message, job_id)	A base class for errors related to arq jobs.
JobNotQueued(job_id)	The job was not successfully queued.
JobNotFound(job_id)	A job cannot be found.
JobResultUnavailable(job_id)	The job's result is unavailable.
ArqMode(value)	Mode configuration for the Arq queue.
JobMetadata(id, name, args, kwargs, ...)	Information about a queued job.
JobResult(id, name, args, kwargs, ...)	The full result of a job, as well as its metadata.
ArqQueue(*[, default_queue_name])	An common interface for working with an arq queue that can be implemented either with a real Redis backend, or an in-memory repository for testing.
RedisArqQueue(pool, *[, default_queue_name])	A distributed queue, based on arq and Redis.
MockArqQueue(*[, default_queue_name])	A mocked queue for testing API services.

Class Inheritance Diagram

safir.database Module

Utility functions for database management.

Functions

create_async_session(engine[, logger, statement])	Create a new async database session.
create_database_engine(url, password, *[, ...])	Create a new async database engine.
create_sync_session(url, password[, logger, ...])	Create a new sync database session.
datetime_from_db()	Add the UTC time zone to a naive datetime from the database.
datetime_to_db()	Strip time zone for storing a datetime in the database.
initialize_database(engine, logger, *, schema)	Create and initialize a new database.

Classes

DatabaseInitializationError

Database initialization failed.

Class Inheritance Diagram

safir.dependencies.arq Module

A FastAPI dependency that supplies a Redis connection for arq.

Classes

ArqDependency()

A FastAPI dependency that maintains a Redis client for enqueing tasks to the worker pool.

Variables

arq_dependency

Singleton instance of ArqDependency that serves as a FastAPI dependency.

Class Inheritance Diagram

safir.dependencies.db_session Module

Manage an async database session.

Classes

DatabaseSessionDependency()

Manages an async per-request SQLAlchemy session.

Variables

db_session_dependency

The dependency that will return the async session proxy.

Class Inheritance Diagram

safir.dependencies.gafaelfawr Module

Gafaelfawr authentication dependencies.

Functions

auth_dependency([x_auth_request_user])	Retrieve authentication information from HTTP headers.
auth_logger_dependency([user, logger])	Logger bound to the authenticated user.

safir.dependencies.http_client Module

HTTP client dependency for FastAPI.

Classes

HTTPClientDependency()

Provides an httpx.AsyncClient as a dependency.

Variables

DEFAULT_HTTP_TIMEOUT	Default timeout (in seconds) for outbound HTTP requests.
http_client_dependency	The dependency that will return the HTTP client.

Class Inheritance Diagram

safir.dependencies.logger Module

Logger dependency for FastAPI.

Provides a structlog logger as a FastAPI dependency. The logger will incorporate information from the request in its bound context.

Classes

LoggerDependency()

Provides a structlog logger configured with request information.

Variables

logger_dependency

The dependency that will return the logger for the current request.

Class Inheritance Diagram

safir.kubernetes Module

Utilities for configuring a Kubernetes client.

Functions

initialize_kubernetes()

Load the Kubernetes configuration.

Variables

annotations

safir.logging Module

Utilities for configuring structlog-based logging.

Functions

add_log_severity(logger, method_name, event_dict)	Add the log level to the event dict as severity.
configure_logging(*, name[, profile, ...])	Configure logging and structlog.
configure_uvicorn_logging([log_level])	Set up logging.

Classes

LogLevel(value)	Python logging level.
Profile(value)	Logging profile for the application.

Variables

logger_name

Name of the configured global logger.

Class Inheritance Diagram

safir.metadata Module

Standardized metadata for Roundtable HTTP services.

Functions

get_metadata(*, package_name, application_name)	Retrieve metadata for the application.
get_project_url(meta, label)	Get a specific URL from a package's project_urls metadata.

Classes

Metadata(**data)

Metadata about a package.

Class Inheritance Diagram

safir.models Module

Standard models for FastAPI applications.

Examples

To reference the ErrorModel model when returning an error message, use code similar to this:

@router.get(
    "/route/{foo}",
    ...,
    responses={404: {"description": "Not found", "model": ErrorModel}},
)
async def route(foo: str) -> None:
    ...
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail=[
            {"loc": ["path", "foo"], "msg": msg, "type": "invalid_foo"},
        ],
    )

Notes

FastAPI does not appear to export its error response model in a usable form, so define a copy of it so that we can reference it in API definitions to generate good documentation.

Classes

ErrorDetail(**data)	The detail of the error message.
ErrorModel(**data)	A structured API error message.

Class Inheritance Diagram

safir.middleware.ivoa Module

Middleware for IVOA services.

Classes

CaseInsensitiveQueryMiddleware(app[, dispatch])

Make query parameter keys all lowercase.

Class Inheritance Diagram

safir.middleware.x_forwarded Module

Update the request based on X-Forwarded-For headers.

Classes

XForwardedMiddleware(app, *[, proxies])

Middleware to update the request based on X-Forwarded-For.

Class Inheritance Diagram

safir.testing.kubernetes Module

Mock Kubernetes API for testing.

Functions

patch_kubernetes()

Replace the Kubernetes API with a mock class.

Classes

MockKubernetesApi()

Mock Kubernetes API for testing.

Class Inheritance Diagram