API reference

safir Module

safir.arq Package

An arq client with a mock for testing.

Functions

build_arq_redis_settings(url, password)

Construct Redis settings for arq.

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])	A 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.
WorkerSettings(functions, redis_settings, ...)	Configuration class for an arq worker.

Class Inheritance Diagram

safir.arq.uws Module

Construction of UWS backend workers.

Functions

build_worker(worker, config, logger)

Construct an arq worker for the provided backend function.

Classes

WorkerConfig(arq_mode, arq_queue_url, ...)	Minimal configuration needed for building a UWS backend worker.
WorkerError(message[, detail, add_traceback])	An error occurred during background task processing.
WorkerErrorType(value[, names, module, ...])	Types of errors that may be reported by a worker.
WorkerFatalError(message[, detail, ...])	Fatal error occurred during worker processing.
WorkerJobInfo(job_id, user, token, timeout)	Metadata about the job that may be useful to the backend.
WorkerResult	A single result from the job.
WorkerTimeoutError(elapsed, timeout)	Transient error occurred during worker processing.
WorkerTransientError(message[, detail, ...])	Transient error occurred during worker processing.
WorkerUsageError(message[, detail, ...])	Parameters sent by the user were invalid.

Variables

T	Type for job parameters.
UWS_QUEUE_NAME	Name of the arq queue for internal UWS messages.

Class Inheritance Diagram

safir.asyncio Package

Utility functions for asyncio code.

Functions

run_with_asyncio(f)

Run the decorated function with asyncio.run.

Classes

AsyncMultiQueue()	An asyncio multiple reader, multiple writer queue.
AsyncMultiQueueError	Invalid sequence of calls when writing to AsyncMultiQueue.

Variables

F	Type variable.
P	Parameter specification variable.

Class Inheritance Diagram

safir.click Package

Click command-line interface support.

Functions

display_help(main, ctx[, topic, subtopic])

Show help for a Click command.

safir.database Package

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.
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.
drop_database(engine, schema)	Drop all tables from the database.
initialize_database(engine, logger, *, schema)	Create and initialize a new database.
is_database_current(engine[, logger, ...])	Check whether the database schema is at the current version.
retry_async_transaction([__func, delay, ...])	Retry if a transaction failed.
run_migrations_offline(metadata, url)	Run Alembic migrations in offline mode.
run_migrations_online(metadata, url, password)	Run Alembic migrations online using an async backend.
stamp_database(config_path)	Mark the database as updated to the head of the given Alembic config.
stamp_database_async(engine[, config_path])	Mark the database as updated to the head of the given Alembic config.
unstamp_database(engine)	Clear the Alembic version from the database.

Classes

AlembicConfigError	The Alembic configuration was missing or invalid.
DatabaseInitializationError	Database initialization failed.

Variables

RetryP	Parameter specification variable.
RetryT	Type variable.

Class Inheritance Diagram

safir.datetime Package

Date and time manipulation utility functions.

Functions

current_datetime(*[, microseconds])	Construct a 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.
parse_timedelta(text)	Parse a string into a datetime.timedelta.

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 enqueuing 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_delegated_token_dependency(...)	Retrieve Gafaelfawr delegated token from HTTP headers.
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.dependencies.metrics Module

Dependencies for metrics functionality.

Classes

EventDependency(event_maker)	Provides EventManager-managed events for apps to publish.
EventMaker()	A blueprint for an event publisher container class.

Variables

E	Generic event maker type.

Class Inheritance Diagram

safir.fastapi Package

Helper code for FastAPI (other than dependencies and middleware).

Functions

client_request_error_handler(request, exc)

Exception handler for exceptions derived from ClientRequestError.

Classes

ClientRequestError(message[, location, ...])

Represents an error in a client request.

Class Inheritance Diagram

safir.gcs Package

Utilities for interacting with Google Cloud Storage.

Classes

SignedURLService(service_account[, lifetime])

Generate signed URLs for Google Cloud Storage blobs.

Class Inheritance Diagram

safir.github Package

GitHub API client factory and Pydantic models.

Classes

GitHubAppClientFactory(*, id, key, name, ...)

Factory for creating GitHub App clients authenticated either as an app or as an installation of that app.

Class Inheritance Diagram

safir.github.models Module

Pydantic models for GitHub v3 REST API resources.

Classes

GitHubRepoOwnerModel	A Pydantic model for the owner field found in repository objects.
GitHubUserModel	A Pydantic model for the user field found in GitHub API resources.
GitHubRepositoryModel	A Pydantic model for the repository field, often found in webhook payloads.
GitHubPullState(value[, names, module, ...])	The state of a GitHub pull request (PR).
GitHubPullRequestModel	A Pydantic model for a GitHub Pull Request.
GitHubBranchCommitModel	A Pydantic model for the commit field found in GitHubBranchModel.
GitHubBranchModel	A Pydantic model for a GitHub branch.
GitHubBlobModel	A Pydantic model for a blob, returned by the GitHub blob endpoint.
GitHubCheckSuiteStatus(value[, names, ...])	The status of a GitHub check suite.
GitHubCheckSuiteConclusion(value[, names, ...])	The conclusion state of a GitHub check suite.
GitHubCheckSuiteModel	A Pydantic model for the check_suite field in a check_suite webhook (GitHubCheckSuiteEventModel).
GitHubCheckRunStatus(value[, names, module, ...])	The check run status.
GitHubCheckRunConclusion(value[, names, ...])	The check run conclusion state.
GitHubCheckRunAnnotationLevel(value[, ...])	The level of a check run output annotation.
GitHubCheckSuiteId	Brief information about a check suite in the GitHubCheckRunModel.
GitHubCheckRunOutput	Check run output report.
GitHubCheckRunPrInfoModel	A Pydantic model of the pull_requests[] items in a check run GitHub API model (GitHubCheckRunModel).
GitHubCheckRunModel	A Pydantic model for the "check_run" field in a check_run webhook payload (GitHubCheckRunEventModel).

Class Inheritance Diagram

safir.github.webhooks Module

Pydantic models for GitHub webhook payloads.

Classes

GitHubAppInstallationModel	A Pydantic model for the installation field found in webhook payloads for GitHub Apps.
GitHubPushEventModel	A Pydantic model for the push event webhook when a commit or tag is pushed.
GitHubAppInstallationEventRepoModel	A pydantic model for repository objects used by GitHubAppInstallationRepositoriesEventModel.
GitHubAppInstallationEventAction(value[, ...])	The action performed on an GitHub App installation webhook (GitHubAppInstallationEventModel).
GitHubAppInstallationEventModel	A Pydantic model for an installation webhook.
GitHubAppInstallationRepositoriesEventAction(value)	The action performed on a GitHub App installation_repositories webhook (GitHubAppInstallationRepositoriesEventModel).
GitHubAppInstallationRepositoriesEventModel	A Pydantic model for a installation_repositories webhook.
GitHubPullRequestEventAction(value[, names, ...])	The action performed on a GitHub pull_request webhook (GitHubPullRequestEventModel).
GitHubPullRequestEventModel	A Pydantic model for a pull_request webhook.
GitHubCheckSuiteEventAction(value[, names, ...])	The action performed in a GitHub check_suite webhook (GitHubCheckSuiteEventModel).
GitHubCheckSuiteEventModel	A Pydantic model for the check_suite webhook payload.
GitHubCheckRunEventAction(value[, names, ...])	The action performed in a GitHub check_run webhook (GitHubCheckRunEventModel).
GitHubCheckRunEventModel	A Pydantic model for the check_run webhook payload.

Class Inheritance Diagram

safir.kafka Package

Classes

AIOKafkaParams	Type for parameters to the constructor of an aiokafka client.
FastStreamBrokerParams	Type for parameters to the constructor of a FastStream broker.
IncompatibleSchemaError	A schema is incompatible with the latest version in the registry.
InvalidAvroNameError	The decalred name or namespace for an Avro schema is not valid.
InvalidMetadataError	The Meta inner class on a model has unexpected values in fields.
KafkaConnectionSettings	Settings for connecting to Kafka.
PlaintextSettings	Subset of settings required for Plaintext auth.
PlaintextSettings	Subset of settings required for Plaintext auth.
SaslMechanism(value[, names, module, ...])	Kafka SASL mechanisms.
SaslPlaintextSettings	Subset of settings required for SASL SSLauth.
SaslSslSettings	Subset of settings required for SASL PLAINTEXT auth.
SecurityProtocol(value[, names, module, ...])	Kafka SASL security protocols.
SslSettings	Subset of settings required for SSL auth.
SslSettings	Subset of settings required for SSL auth.
PydanticSchemaManager(registry[, suffix, logger])	A manager for schemas that are represented as Pydantic models in Python, and translated into Avro for the Confluent Schema Registry.
SchemaInfo(schema, schema_id, subject)	Schema and registry metadata.
SchemaManagerSettings	Settings for constructing a PydanticSchemaManager.
SchemaManagerSettings	Settings for constructing a PydanticSchemaManager.
SchemaRegistryClientParams	Kwargs used to construct an AsyncSchemaRegistryClient.
SchemaRegistryClientParams	Kwargs used to construct an AsyncSchemaRegistryClient.
Compatibility(value[, names, module, ...])	Schema registry compatibility types.
UnknownDeserializeError	The schema registry client returns None when deserializing.
UnknownSchemaError(data, subject)	A schema is not managed by the Registry, and therefore cannot be serialized into a native Python object.

Class Inheritance Diagram

safir.kubernetes Package

Utilities for configuring a Kubernetes client.

Functions

initialize_kubernetes()

Load the Kubernetes configuration.

safir.logging Package

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_alembic_logging([log_level])	Set up logging for Alembic.
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.

Class Inheritance Diagram

safir.metadata Package

Standardized metadata for Roundtable and Phalanx 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

Metadata about a package.

Class Inheritance Diagram

safir.metrics Package

Functions

metrics_configuration_factory()

Choose an appropriate metrics configuration based on the environment.

Classes

BaseMetricsConfiguration	Metrics configuration, including the required Kafka configuration.
DisabledMetricsConfiguration	Metrics configuration when metrics reporting is disabled.
DuplicateEventError(name)	Two publishers were registered with the same name.
EventsConfiguration	Configuration for emitting events.
EventManager(topic[, logger])	Interface for a client for publishing application metrics events.
EventManagerUnintializedError	An attempt to create a publisher after manager has been initialized.
EventMetadata	Common fields for all metrics events.
EventPayload	All event payloads should inherit from this.
EventPublisher(application, event_class)	Interface for event publishers.
KafkaEventManager(*, application, ...[, ...])	A tool for publishing application metrics events.
KafkaEventPublisher(*, application, manager, ...)	Publishes one type of event.
KafkaMetricsConfiguration	Metrics configuration when enabled, including Kafka configuration.
KafkaTopicError(topic)	A topic does not exist in Kafka, or we don't have access to it.
KafkaTopicError(topic)	A topic does not exist in Kafka, or we don't have access to it.
NoopEventManager(application, topic_prefix)	An event manager that creates publishers that quietly do nothing.
NoopEventPublisher(application, event_class, ...)	Event publisher that quietly does nothing.

Variables

MetricsConfiguration

Represent a PEP 604 union type

Class Inheritance Diagram

safir.models Package

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	The detail of the error message.
ErrorLocation(value[, names, module, ...])	Possible locations for an error.
ErrorModel	A structured API error message.

Class Inheritance Diagram

safir.middleware.ivoa Module

Middleware for IVOA services.

Classes

CaseInsensitiveQueryMiddleware(app)

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])

ASGI middleware to update the request based on X-Forwarded-For.

Class Inheritance Diagram

safir.pydantic Package

Utilities for Pydantic models.

Functions

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

Classes

CamelCaseModel

pydantic.BaseModel configured to accept camel-case input.

Class Inheritance Diagram

safir.redis Package

Redis database support.

Classes

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

Class Inheritance Diagram

safir.slack.blockkit Module

Slack Block Kit message models.

Classes

SlackBaseBlock	Base class for any Slack Block Kit block.
SlackBaseField	Base class for Slack Block Kit blocks for the fields section.
SlackCodeBlock	A component of a Slack message with a heading and a code block.
SlackCodeField	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	Message to post to Slack.
SlackTextBlock	A component of a Slack message with a heading and a text body.
SlackTextField	One field in a Slack message with a heading and text body.
SlackWebException(message, *[, failed_at, ...])	Parent class of exceptions arising from HTTPX failures.

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

safir.testing.uws Module

Mock UWS job executor for testing.

Classes

MockUWSJobRunner(config, arq_queue)

Simulate execution of jobs with a mock queue.

Class Inheritance Diagram

safir.uws Package

Support library for writing UWS-enabled services.

Functions

uws_post_params_dependency(request)

Parse POST parameters.

Classes

DatabaseSchemaError	Some problem was detected in the UWS database schema.
ErrorCode(value[, names, module, qualname, ...])	Possible error codes in text/plain responses.
MultiValuedParameterError(message)	Multiple values not allowed for this parameter.
ParameterError(message[, detail])	Unsupported value passed to a parameter.
ParameterParseError(message, params)	UWS job parameters could not be parsed.
ParametersModel	Defines the interface for a model suitable for job parameters.
UWSApplication(config)	Glue between a FastAPI application and the UWS implementation.
UWSAppSettings	Settings common to all applications using the UWS library.
UWSConfig(arq_mode, arq_redis_settings, ...)	Configuration for the UWS service.
UWSError(error_code, message[, detail])	An error with an associated error code.
UWSJob(job_id, message_id, owner, phase, ...)	Represents a single UWS job.
UWSJobError(error_type, error_code, message)	Failure information about a job.
UWSJobParameter(parameter_id, value)	An input parameter to the job.
UWSJobResult(result_id, url[, size, mime_type])	A single result from the job.
UWSRoute(dependency, summary[, description])	Defines a FastAPI dependency to get the UWS job parameters.
UWSSchemaBase(**kwargs)	SQLAlchemy declarative base for the UWS database schema.
UsageError(message[, detail])	Invalid parameters were passed to a UWS API.

Class Inheritance Diagram