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[, names, module, qualname, ...])	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.asyncio Module

Utility functions for asyncio code.

Functions

run_with_asyncio(f)

Run the decorated function with asyncio.run.

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(time)	Add the UTC time zone to a naive datetime from the database.
datetime_to_db(time)	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.datetime Module

Date and time manipulation utility functions.

Functions

current_datetime(*[, microseconds])	Construct a ~datetime.datetime for the current time.
format_datetime_for_logging(timestamp)	Format a datetime for logging and human readabilty.
isodatetime(timestamp)	Format a timestamp in UTC in a standard ISO date format.
parse_isodatetime(time_string)	Parse a string in a standard ISO date format.

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.gcs Module

Utilities for interacting with Google Cloud Storage.

Classes

SignedURLService(service_account[, lifetime])

Generate signed URLs for Google Cloud Storage blobs.

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[, names, module, qualname, ...])	Python logging level.
Profile(value[, names, module, qualname, ...])	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.

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.
ErrorLocation(value[, names, module, ...])	Possible locations for an error.
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.pydantic Module

Utility functions for Pydantic models.

Functions

normalize_datetime(v)	Pydantic validator for datetime fields.
normalize_isodatetime(v)	Pydantic validator for datetime fields in ISO format.
to_camel_case(string)	Convert a string to camel case.
validate_exactly_one_of(*settings)	Generate a validator imposing a one and only one constraint.

Classes

CamelCaseModel(**data)

pydantic.BaseModel configured to accept camel-case input.

Class Inheritance Diagram

safir.redis Module

Redis database support.

Classes

PydanticRedisStorage(*, datatype, redis[, ...])	JSON-serialized encrypted storage in Redis.
EncryptedPydanticRedisStorage(*, datatype, ...)	A Pydantic-based Redis store that encrypts data.
DeserializeError(msg, key, error)	Raised when a stored Pydantic object in Redis cannot be decoded (and possibly decrypted) or deserialized.

Variables

S	Type variable of Pydantic object being stored in PydanticRedisStorage.

Class Inheritance Diagram

safir.slack.blockkit Module

Slack Block Kit message models.

Classes

SlackBaseBlock(**data)	Base class for any Slack Block Kit block.
SlackBaseField(**data)	Base class for Slack Block Kit blocks for the fields section.
SlackCodeBlock(**data)	A component of a Slack message with a heading and a code block.
SlackCodeField(**data)	An attachment in a Slack message with a heading and text body.
SlackException(message[, user, failed_at])	Parent class of exceptions that can be reported to Slack.
SlackMessage(**data)	Message to post to Slack.
SlackTextBlock(**data)	A component of a Slack message with a heading and a text body.
SlackTextField(**data)	One field in a Slack message with a heading and text body.

Class Inheritance Diagram

safir.slack.webhook Module

Send messages to Slack.

Classes

SlackIgnoredException	Parent class for exceptions that should not be reported to Slack.
SlackRouteErrorHandler(path, endpoint, *[, ...])	Custom fastapi.routing.APIRoute that reports exceptions to Slack.
SlackWebhookClient(hook_url, application, logger)	Send messages to a Slack webhook.

Class Inheritance Diagram

safir.testing.gcs Module

Mock Google Cloud Storage API for testing.

Functions

patch_google_storage(*[, ...])

Replace the Google Cloud Storage API with a mock class.

Classes

MockBlob(name[, expected_expiration])	Mock version of google.cloud.storage.blob.Blob.
MockBucket(bucket_name[, ...])	Mock version of google.cloud.storage.bucket.Bucket.
MockStorageClient([expected_expiration, ...])	Mock version of google.cloud.storage.Client.

Class Inheritance Diagram

safir.testing.kubernetes Module

Mock Kubernetes API for testing.

Functions

patch_kubernetes()	Replace the Kubernetes API with a mock class.
strip_none(model)	Strip None values from a serialized Kubernetes object.

Classes

MockKubernetesApi()

Mock Kubernetes API for testing.

Class Inheritance Diagram

safir.testing.slack Module

Mock Slack server for testing Slack messaging.

Functions

mock_slack_webhook(hook_url, respx_mock)

Set up a mocked Slack server.

Classes

MockSlackWebhook(url)

Represents a Slack incoming webhook and remembers what was posted.

Class Inheritance Diagram

safir.testing.uvicorn Module

Utiility functions for managing an external Uvicorn test process.

Normally, ASGI apps are tested via the built-in support in httpx for running an ASGI app directly. However, sometimes the app has to be spawned in a separate process so that it can be accessed over HTTP, such as when testing it with Selenium or when testing Uvicorn integration. This module provides utility functions to aid with that test setup.

Functions

spawn_uvicorn(*, working_directory[, app, ...])

Spawn an ASGI app as a separate Uvicorn process.

Classes

ServerNotListeningError	Timeout waiting for the server to start listening.
UvicornProcess(process, url)	Properties of the running Uvicorn service.

Class Inheritance Diagram