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)
_cli_kebab_case (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"
      },
      "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"
      },
      "wobbly_url": {
         "description": "URL to Wobbly UWS job tracking API",
         "format": "uri",
         "maxLength": 2083,
         "minLength": 1,
         "title": "Wobbly URL",
         "type": "string"
      }
   },
   "$defs": {
      "ArqMode": {
         "description": "Mode configuration for the Arq queue.",
         "enum": [
            "production",
            "test"
         ],
         "title": "ArqMode",
         "type": "string"
      }
   },
   "additionalProperties": false,
   "required": [
      "arq_queue_url",
      "service_account",
      "storage_url",
      "wobbly_url"
   ]
}

Fields:

arq_mode (safir.arq._models.ArqMode)
arq_queue_password (pydantic.types.SecretStr | None)
arq_queue_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)
wobbly_url (pydantic.networks.HttpUrl)

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 0x7ff01c54ed40>

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 0x7ff01c54f1a0>
json_schema_input_type = PydanticUndefined
return_type = <class ‘float’>
when_used = json

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

Constraints:

func = <function <lambda> at 0x7ff01c54f060>
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 0x7ff01c54f060>
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 0x7ff01c54f1a0>
json_schema_input_type = PydanticUndefined
return_type = <class ‘float’>
when_used = json

field wobbly_url: HttpUrl [Required]¶: URL to Wobbly UWS job tracking API

build_uws_config(*, async_post_route, job_summary_type, 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.

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 Pydantic model representing the job parameters.
job_summary_type (type[JobSummary]) – Type representing the XML job summary type, qualified with an appropriate subclass of Parameters. That subclass should be the same as that returned by the to_xml_model method of parameters_type.
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 Pydantic model 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 Pydantic model 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, Job], 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 | None, Job], timedelta | None] | 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.

Returns:

UWS configuration.

Return type:

UWSConfig

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.