Creating a new UWS service

To create a new service that uses the Safir UWS library for its API, first create a new FastAPI Safir application. The easiest way to do this is to follow the instructions in Creating an app from the template. Select the UWS flavor.

Then, flesh out the application by following these steps:

  1. Define the parameter models

  2. Define the API parameters

  3. Write the backend worker

  4. Write the test suite

If you use the template and select the UWS flavor, all of the steps below will be done for you and you can skip the rest of this page. Read on if you’re curious about what the UWS flavor sets up, or if you’re converting an already-existing FastAPI application.

Add UWS configuration options

UWS applications have several standard configuration options that you will want to include in your application’s overall configuration. You will also need to add a method to creete the UWSConfig object from your application configuration and to create a global UWSApplication object. These need to be added to config.py.

First, add additional configuration settings for UWS to Config by changing the class to inherit from UWSAppSettings. This will add standard configuration options most services will need and provide helper methods and properties.

config.py
from safir.uws import UWSAppSettings


class Config(UWSAppSettings): ...

Second, add a property to Config that returns the UWS configuration. For some of these settings, you won’t know the values yet.

You will be able to fill in the values of job_summary_type and parameters_type after reading Define job parameter models. You will be able to fill in the values of async_post_route and optionally sync_get_route and sync_post_route after reading Defining service inputs. You will be able to fill in the value of worker after reading Write the backend worker. For now, you can just insert placeholder values.

config.py
from safir.uws import UWSAppSettings, UWSConfig, UWSRoute


class Config(UWSAppSettings):
    ...

    @property
    def uws_config(self) -> UWSConfig:
        return self.build_uws_config(
            async_post_route=UWSRoute(...),
            job_summary_type=...,
            parameters_type=...,
            sync_get_route=UWSRoute(...),
            sync_post_route=UWSRoute(...),
            worker=...,
        )

See UWSAppSettings.build_uws_config for all of the possible settings.

Third, at the bottom of config.py, create the UWSApplication object and store it in uws, which should be an exported symbol (listed in __all__).

config.py
from safir.uws import UWSApplication

...

config = Config()
"""Configuration for example."""

uws = UWSApplication(config.uws_config)
"""The UWS application for this service."""

Set up the FastAPI application

The Safir UWS library must be initialized when the application starts, and requires some additional FastAPI middleware and error handlers. These need to be added to main.py.

First, initialize and shut down the UWS application in the lifespan function:

main.py
from safir.dependencies.http_client import http_client_dependency

from .config import uws


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    await uws.initialize_fastapi()
    yield
    await uws.shutdown_fastapi()
    await http_client_dependency.aclose()

Second, install the UWS routes into the external router before including it in the application:

main.py
# Attach the routers.
app.include_router(internal_router)
uws.install_handlers(external_router)
app.include_router(external_router, prefix=f"{config.path_prefix}")

Third, install the UWS middleware and error handlers.

main.py
# Add middleware.
app.add_middleware(XForwardedMiddleware)
uws.install_middleware(app)

# Install error handlers.
uws.install_error_handlers(app)

Create the arq worker for database updates

Your application will have two separate arq worker pods, one to do the actual work of your application and one to handle database updates and state tracking (via API calls to Wobbly). The code for the second worker is part of the UWS library, but you have to add a small amount of code to enable it and attach it to your application configuration.

Create a subdirectory named workers in the source for your application with an empty workers/__init__.py file. Then, create workers/uws.py with the following contents:

workers/uws.py
import structlog
from safir.logging import configure_logging

from ..config import config, uws


configure_logging(
    name="example", profile=config.profile, log_level=config.log_level
)

WorkerSettings = uws.build_worker(structlog.get_logger("example"))
"""arq configuration for the UWS database worker."""

Once again, replace example with the name of your application.

Next steps

Now that you have set up the basic structure of your application, you can move on to the substantive parts.