UWSAppSettings

pydantic settings safir.uws.UWSAppSettings

Settings common to all applications using the UWS library.

The Config class for an application should inherit from this class to get the standard UWS application settings.

Parameters:

• _case_sensitive (bool | None, default: None)

• _nested_model_default_partial_update (bool | None, default: None)

• _env_prefix (str | None, default: None)

• _env_file (Union[Path, str, Sequence[Union[Path, str]], None], default: PosixPath('.'))

• _env_file_encoding (str | None, default: None)

• _env_ignore_empty (bool | None, default: None)

• _env_nested_delimiter (str | None, default: None)

• _env_parse_none_str (str | None, default: None)

• _env_parse_enums (bool | None, default: None)

• _cli_prog_name (str | None, default: None)

• _cli_parse_args (bool | list[str] | tuple[str, ...] | None, default: None)

• _cli_settings_source (Optional[CliSettingsSource[Any]], default: None)

• _cli_parse_none_str (str | None, default: None)

• _cli_hide_none_type (bool | None, default: None)

• _cli_avoid_json (bool | None, default: None)

• _cli_enforce_required (bool | None, default: None)

• _cli_use_class_docs_for_groups (bool | None, default: None)

• _cli_exit_on_error (bool | None, default: None)

• _cli_prefix (str | None, default: None)

• _cli_flag_prefix_char (str | None, default: None)

• _cli_implicit_flags (bool | None, default: None)

• _cli_ignore_unknown_args (bool | None, default: None)

• _secrets_dir (Union[Path, str, Sequence[Union[Path, str]], None], default: None)

• values (Any)

Show JSON schema

{
   "title": "UWSAppSettings",
   "description": "Settings common to all applications using the UWS library.\n\nThe ``Config`` class for an application should inherit from this class to\nget the standard UWS application settings.",
   "type": "object",
   "properties": {
      "arq_mode": {
         "$ref": "#/$defs/ArqMode",
         "default": "production",
         "description": "This will always be production outside the test suite",
         "title": "arq operation mode"
      },
      "arq_queue_url": {
         "description": "DSN of Redis server to use for the arq queue",
         "format": "uri",
         "minLength": 1,
         "title": "arq Redis DSN",
         "type": "string"
      },
      "arq_queue_password": {
         "anyOf": [
            {
               "format": "password",
               "type": "string",
               "writeOnly": true
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Password of Redis server to use for the arq queue",
         "title": "Password for arq Redis server"
      },
      "database_url": {
         "description": "DSN of PostgreSQL database for UWS job tracking",
         "format": "uri",
         "minLength": 1,
         "title": "PostgreSQL DSN",
         "type": "string"
      },
      "database_password": {
         "anyOf": [
            {
               "format": "password",
               "type": "string",
               "writeOnly": true
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Password for UWS job database"
      },
      "grace_period": {
         "default": "PT30S",
         "description": "How long to wait for a job to finish on shutdown before canceling it",
         "format": "duration",
         "title": "Grace period for jobs",
         "type": "string"
      },
      "lifetime": {
         "default": "P7D",
         "format": "duration",
         "title": "Lifetime of job results",
         "type": "string"
      },
      "service_account": {
         "description": "Email of the service account to use for signed URLs of results. The default credentials that the application frontend runs with must have the ``roles/iam.serviceAccountTokenCreator`` role on the service account with this email.",
         "title": "Service account for URL signing",
         "type": "string"
      },
      "storage_url": {
         "description": "Must be a ``gs`` or ``s3`` URL pointing to a Google Cloud Storage bucket that is writable by the backend and readable by the frontend.",
         "title": "Root URL for cutout results",
         "type": "string"
      },
      "sync_timeout": {
         "default": "PT1M",
         "format": "duration",
         "title": "Timeout for sync requests",
         "type": "string"
      },
      "timeout": {
         "default": "PT10M",
         "description": "Must be given as a number of seconds as a string or integer",
         "format": "duration",
         "title": "Job timeout in seconds",
         "type": "string"
      }
   },
   "$defs": {
      "ArqMode": {
         "description": "Mode configuration for the Arq queue.",
         "enum": [
            "production",
            "test"
         ],
         "title": "ArqMode",
         "type": "string"
      }
   },
   "additionalProperties": false,
   "required": [
      "arq_queue_url",
      "database_url",
      "service_account",
      "storage_url"
   ]
}

Fields:

• arq_mode (safir.arq._models.ArqMode)

• arq_queue_password (pydantic.types.SecretStr | None)

• arq_queue_url (pydantic_core._pydantic_core.Url)

• database_password (pydantic.types.SecretStr | None)

• database_url (pydantic_core._pydantic_core.Url)

• grace_period (datetime.timedelta)

• lifetime (datetime.timedelta)

• service_account (str)

• storage_url (str)

• sync_timeout (datetime.timedelta)

• timeout (datetime.timedelta)

field arq_mode: ArqMode = ArqMode.production

This will always be production outside the test suite

field arq_queue_password: SecretStr | None = None

Password of Redis server to use for the arq queue

field arq_queue_url: EnvRedisDsn [Required]

DSN of Redis server to use for the arq queue

Constraints:

• allowed_schemes = [‘redis’]

• default_host = localhost

• default_port = 6379

• default_path = /0

• func = <function _validate_env_redis_dsn at 0x7fb3ff95f880>

field database_password: SecretStr | None = None

field database_url: EnvAsyncPostgresDsn [Required]

DSN of PostgreSQL database for UWS job tracking

Constraints:

• allowed_schemes = [‘postgresql’, ‘postgresql+asyncpg’]

• host_required = True

• func = <function _validate_env_async_postgres_dsn at 0x7fb3ff95f420>

field grace_period: SecondsTimedelta = datetime.timedelta(seconds=30)

How long to wait for a job to finish on shutdown before canceling it

Constraints:

• func = <function <lambda> at 0x7fb40e244ea0>

• json_schema_input_type = PydanticUndefined

• return_type = <class ‘float’>

• when_used = json

field lifetime: HumanTimedelta = datetime.timedelta(days=7)

Constraints:

• func = <function <lambda> at 0x7fb40e244d60>

• json_schema_input_type = PydanticUndefined

• return_type = <class ‘float’>

• when_used = json

field service_account: str [Required]

Email of the service account to use for signed URLs of results. The default credentials that the application frontend runs with must have the roles/iam.serviceAccountTokenCreator role on the service account with this email.

field storage_url: str [Required]

Must be a gs or s3 URL pointing to a Google Cloud Storage bucket that is writable by the backend and readable by the frontend.

field sync_timeout: HumanTimedelta = datetime.timedelta(seconds=60)

Constraints:

• func = <function <lambda> at 0x7fb40e244d60>

• json_schema_input_type = PydanticUndefined

• return_type = <class ‘float’>

• when_used = json

field timeout: SecondsTimedelta = datetime.timedelta(seconds=600)

Must be given as a number of seconds as a string or integer

Constraints:

• func = <function <lambda> at 0x7fb40e244ea0>

• json_schema_input_type = PydanticUndefined

• return_type = <class ‘float’>

• when_used = json

build_uws_config(*, async_post_route, parameters_type, slack_webhook=None, sync_get_route=None, sync_post_route=None, url_lifetime=datetime.timedelta(seconds=900), validate_destruction=None, validate_execution_duration=None, wait_timeout=datetime.timedelta(seconds=60), worker)

Construct a UWSConfig object from the application configuration.

This helper method can be used by application Config classes to help build the UWSConfig object corresponding to the application configuration. Its parameters are the additional settings accepted by the UWS library that are not part of the UWSAppSettings model.

Returns:

UWS configuration.

Return type:

UWSConfig

Parameters:

• async_post_route (UWSRoute) – Route configuration for job parameters for an async job via POST. The FastAPI dependency included in this object should expect POST parameters and return a list of UWSJobParameter objects representing the job parameters.

• parameters_type (type[ParametersModel]) – Type representing the job parameters. This will be used to validate parameters and to parse them before passing them to the worker.

• slack_webhook (SecretStr | None, default: None) – Slack incoming webhook for reporting errors.

• sync_get_route (UWSRoute | None, default: None) – Route configuration for creating a sync job via GET. The FastAPI dependency included in this object should expect GET parameters and return a list of UWSJobParameter objects representing the job parameters. If None, no route to create a job via sync GET will be created.

• sync_post_route (UWSRoute | None, default: None) – Route configuration for creating a sync job via POST. The FastAPI dependency included in this object should expect POST parameters and return a list of UWSJobParameter objects representing the job parameters. If None, no route to create a job via sync POST will be created.

• url_lifetime (timedelta, default: datetime.timedelta(seconds=900)) – How long result URLs should be valid for.

• validate_destruction (Callable[[datetime, UWSJob], datetime] | None, default: None) – Validate a new destruction time for a job. If provided, called with the requested destruction time and the current job record and should return the new destruction time. Otherwise, any destruction time before the configured maximum lifetime will be allowed.

• validate_execution_duration (Callable[[timedelta, UWSJob], timedelta] | None, default: None) – Validate a new execution duration for a job. If provided, called with the requested execution duration and the current job record and should return the new execution duration time. Otherwise, the execution duration may not be changed.

• wait_timeout (timedelta, default: datetime.timedelta(seconds=60)) – Maximum time a client can wait for a job change.

• worker (str) – Name of the backend worker to call to execute a job.

Examples

Normally, this method is used from a property method that returns the UWS configuration, such as the following example for a cutout service:

@property
def uws_config(self) -> UWSConfig:
    return self.build_uws_config(
        async_post_route=UWSRoute(
            dependency=post_params_dependency,
            summary="Create async cutout job",
        ),
        parameters_type=CutoutParameters,
        slack_webhook=self.slack_webhook,
        sync_get_route=UWSRoute(
            dependency=get_params_dependency,
            summary="Synchronous cutout",
        ),
        sync_post_route=UWSRoute(
            dependency=post_params_dependency,
            summary="Synchronous cutout",
        ),
        worker="cutout",
    )

property arq_redis_settings: RedisSettings

Redis settings for arq.