Settings

Configuration settings for DiracX services and components. See also environment variables and dev environment variables.

settings

Settings for the core services.

Attributes

T = TypeVar('T') module-attribute

TokenSigningKeyStore = Annotated[_TokenSigningKeyStore, BeforeValidator(_maybe_load_keys_from_file)] module-attribute

LocalFileUrl = Annotated[FileUrl, BeforeValidator(_apply_default_scheme)] module-attribute

Classes

SqlalchemyDsn

Bases: AnyUrl

Source code in diracx-core/src/diracx/core/settings.py

class SqlalchemyDsn(AnyUrl):
    _constraints = UrlConstraints(
        allowed_schemes=[
            "sqlite+aiosqlite",
            "mysql+aiomysql",
            # The real scheme is with an underscore, (oracle+oracledb_async)
            # but pydantic does not validate it, so we use this hack
            "oracle+oracledb-async",
        ]
    )

FernetKey

Bases: SecretStr

Source code in diracx-core/src/diracx/core/settings.py

class FernetKey(SecretStr):
    fernet: Fernet

    def __init__(self, data: str):
        super().__init__(data)
        self.fernet = Fernet(self.get_secret_value())

Attributes

fernet = Fernet(self.get_secret_value()) instance-attribute

ServiceSettingsBase

Bases: BaseSettings

Source code in diracx-core/src/diracx/core/settings.py

class ServiceSettingsBase(BaseSettings):
    model_config = SettingsConfigDict(frozen=True)

    @classmethod
    def create(cls) -> Self:
        raise NotImplementedError("This should never be called")

    @contextlib.asynccontextmanager
    async def lifetime_function(self) -> AsyncIterator[None]:
        """A context manager that can be used to run code at startup and shutdown."""
        yield

Attributes

model_config = SettingsConfigDict(frozen=True) class-attribute instance-attribute

Functions

create() classmethod

Source code in diracx-core/src/diracx/core/settings.py

@classmethod
def create(cls) -> Self:
    raise NotImplementedError("This should never be called")

lifetime_function() async

A context manager that can be used to run code at startup and shutdown.

Source code in diracx-core/src/diracx/core/settings.py

@contextlib.asynccontextmanager
async def lifetime_function(self) -> AsyncIterator[None]:
    """A context manager that can be used to run code at startup and shutdown."""
    yield

DevelopmentSettings

Bases: ServiceSettingsBase

Settings for the Development Configuration that can influence run time.

Source code in diracx-core/src/diracx/core/settings.py

class DevelopmentSettings(ServiceSettingsBase):
    """Settings for the Development Configuration that can influence run time."""

    model_config = SettingsConfigDict(
        env_prefix="DIRACX_DEV_", use_attribute_docstrings=True
    )

    crash_on_missed_access_policy: bool = False
    """When set to true (only for demo/CI), crash if an access policy isn't called.

    This is useful for development and testing to ensure all endpoints have proper
    access control policies defined.
    """

    @classmethod
    def create(cls) -> Self:
        return cls()

Attributes

model_config = SettingsConfigDict(env_prefix='DIRACX_DEV_', use_attribute_docstrings=True) class-attribute instance-attribute

crash_on_missed_access_policy = False class-attribute instance-attribute

When set to true (only for demo/CI), crash if an access policy isn't called.

This is useful for development and testing to ensure all endpoints have proper access control policies defined.

Functions

create() classmethod

Source code in diracx-core/src/diracx/core/settings.py

@classmethod
def create(cls) -> Self:
    return cls()

AuthSettings

Bases: ServiceSettingsBase

Settings for the authentication service.

Source code in diracx-core/src/diracx/core/settings.py

class AuthSettings(ServiceSettingsBase):
    """Settings for the authentication service."""

    model_config = SettingsConfigDict(
        env_prefix="DIRACX_SERVICE_AUTH_", use_attribute_docstrings=True
    )

    dirac_client_id: str = "myDIRACClientID"
    """OAuth2 client identifier for DIRAC services.

    This should match the client ID registered with the identity provider.
    """

    allowed_redirects: list[str] = []
    """List of allowed redirect URLs for OAuth2 authorization flow.

    These URLs must be pre-registered and should match the redirect URIs
    configured in the OAuth2 client registration.
    Example: ["http://localhost:8000/docs/oauth2-redirect"]
    """

    device_flow_expiration_seconds: int = 600
    """Expiration time in seconds for device flow authorization requests.

    After this time, the device code becomes invalid and users must restart
    the device flow process. Default: 10 minutes.
    """

    authorization_flow_expiration_seconds: int = 300
    """Expiration time in seconds for authorization code flow.

    The time window during which the authorization code remains valid
    before it must be exchanged for tokens. Default: 5 minutes.
    """

    state_key: FernetKey
    """Encryption key used to encrypt/decrypt the state parameter passed to the IAM.

    This key ensures the integrity and confidentiality of state information
    during OAuth2 flows. Must be a valid Fernet key.
    """

    token_issuer: str
    """The issuer identifier for JWT tokens.

    This should be a URI that uniquely identifies the token issuer and
    matches the 'iss' claim in issued JWT tokens.
    """

    token_keystore: TokenSigningKeyStore
    """Keystore containing the cryptographic keys used for signing JWT tokens.

    This includes both public and private keys for token signature
    generation and verification.
    """

    token_allowed_algorithms: list[str] = ["RS256", "EdDSA"]  # noqa: S105
    """List of allowed cryptographic algorithms for JWT token signing.

    Supported algorithms include RS256 (RSA with SHA-256) and EdDSA
    (Edwards-curve Digital Signature Algorithm). Default: ["RS256", "EdDSA"]
    """

    access_token_expire_minutes: int = 20
    """Expiration time in minutes for access tokens.

    After this duration, access tokens become invalid and must be refreshed
    or re-obtained. Default: 20 minutes.
    """

    refresh_token_expire_minutes: int = 60
    """Expiration time in minutes for refresh tokens.

    The maximum lifetime of refresh tokens before they must be re-issued
    through a new authentication flow. Default: 60 minutes.
    """

    available_properties: set[SecurityProperty] = Field(
        default_factory=SecurityProperty.available_properties
    )
    """Set of security properties available in this DIRAC installation.

    These properties define various authorization capabilities and are used
    for access control decisions. Defaults to all available security properties.
    """

Attributes

model_config = SettingsConfigDict(env_prefix='DIRACX_SERVICE_AUTH_', use_attribute_docstrings=True) class-attribute instance-attribute

dirac_client_id = 'myDIRACClientID' class-attribute instance-attribute

OAuth2 client identifier for DIRAC services.

This should match the client ID registered with the identity provider.

allowed_redirects = [] class-attribute instance-attribute

List of allowed redirect URLs for OAuth2 authorization flow.

These URLs must be pre-registered and should match the redirect URIs configured in the OAuth2 client registration. Example: ["http://localhost:8000/docs/oauth2-redirect"]

device_flow_expiration_seconds = 600 class-attribute instance-attribute

Expiration time in seconds for device flow authorization requests.

After this time, the device code becomes invalid and users must restart the device flow process. Default: 10 minutes.

authorization_flow_expiration_seconds = 300 class-attribute instance-attribute

Expiration time in seconds for authorization code flow.

The time window during which the authorization code remains valid before it must be exchanged for tokens. Default: 5 minutes.

state_key instance-attribute

Encryption key used to encrypt/decrypt the state parameter passed to the IAM.

This key ensures the integrity and confidentiality of state information during OAuth2 flows. Must be a valid Fernet key.

token_issuer instance-attribute

The issuer identifier for JWT tokens.

This should be a URI that uniquely identifies the token issuer and matches the 'iss' claim in issued JWT tokens.

token_keystore instance-attribute

Keystore containing the cryptographic keys used for signing JWT tokens.

This includes both public and private keys for token signature generation and verification.

token_allowed_algorithms = ['RS256', 'EdDSA'] class-attribute instance-attribute

List of allowed cryptographic algorithms for JWT token signing.

Supported algorithms include RS256 (RSA with SHA-256) and EdDSA (Edwards-curve Digital Signature Algorithm). Default: ["RS256", "EdDSA"]

access_token_expire_minutes = 20 class-attribute instance-attribute

Expiration time in minutes for access tokens.

After this duration, access tokens become invalid and must be refreshed or re-obtained. Default: 20 minutes.

refresh_token_expire_minutes = 60 class-attribute instance-attribute

Expiration time in minutes for refresh tokens.

The maximum lifetime of refresh tokens before they must be re-issued through a new authentication flow. Default: 60 minutes.

available_properties = Field(default_factory=(SecurityProperty.available_properties)) class-attribute instance-attribute

Set of security properties available in this DIRAC installation.

These properties define various authorization capabilities and are used for access control decisions. Defaults to all available security properties.

SandboxStoreSettings

Bases: ServiceSettingsBase

Settings for the sandbox store.

Source code in diracx-core/src/diracx/core/settings.py

class SandboxStoreSettings(ServiceSettingsBase):
    """Settings for the sandbox store."""

    model_config = SettingsConfigDict(
        env_prefix="DIRACX_SANDBOX_STORE_", use_attribute_docstrings=True
    )

    bucket_name: str
    """Name of the S3 bucket used for storing job sandboxes.

    This bucket will contain input and output sandbox files for DIRAC jobs.
    The bucket must exist or auto_create_bucket must be enabled.
    """

    s3_client_kwargs: dict[str, str]
    """Configuration parameters passed to the S3 client."""

    auto_create_bucket: bool = False
    """Whether to automatically create the S3 bucket if it doesn't exist."""

    url_validity_seconds: int = 5 * 60
    """Validity duration in seconds for pre-signed S3 URLs.

    This determines how long generated download/upload URLs remain valid
    before expiring. Default: 300 seconds (5 minutes).
    """

    se_name: str = "SandboxSE"
    """Logical name of the Storage Element for the sandbox store.

    This name is used within DIRAC to refer to this sandbox storage
    endpoint in job descriptions and file catalogs.
    """
    _client: S3Client = PrivateAttr()

    @contextlib.asynccontextmanager
    async def lifetime_function(self) -> AsyncIterator[None]:
        async with get_session().create_client(
            "s3",
            **self.s3_client_kwargs,
            config=Config(signature_version="v4"),
        ) as self._client:  # type: ignore
            if not await s3_bucket_exists(self._client, self.bucket_name):
                if not self.auto_create_bucket:
                    raise ValueError(
                        f"Bucket {self.bucket_name} does not exist and auto_create_bucket is disabled"
                    )
                try:
                    await self._client.create_bucket(Bucket=self.bucket_name)
                except ClientError as e:
                    raise ValueError(
                        f"Failed to create bucket {self.bucket_name}"
                    ) from e

            yield

    @property
    def s3_client(self) -> S3Client:
        if self._client is None:
            raise RuntimeError("S3 client accessed before lifetime function")
        return self._client

Attributes

model_config = SettingsConfigDict(env_prefix='DIRACX_SANDBOX_STORE_', use_attribute_docstrings=True) class-attribute instance-attribute

bucket_name instance-attribute

Name of the S3 bucket used for storing job sandboxes.

This bucket will contain input and output sandbox files for DIRAC jobs. The bucket must exist or auto_create_bucket must be enabled.

s3_client_kwargs instance-attribute

Configuration parameters passed to the S3 client.

auto_create_bucket = False class-attribute instance-attribute

Whether to automatically create the S3 bucket if it doesn't exist.

url_validity_seconds = 5 * 60 class-attribute instance-attribute

Validity duration in seconds for pre-signed S3 URLs.

This determines how long generated download/upload URLs remain valid before expiring. Default: 300 seconds (5 minutes).

se_name = 'SandboxSE' class-attribute instance-attribute

Logical name of the Storage Element for the sandbox store.

This name is used within DIRAC to refer to this sandbox storage endpoint in job descriptions and file catalogs.

s3_client property

Functions

lifetime_function() async

Source code in diracx-core/src/diracx/core/settings.py

@contextlib.asynccontextmanager
async def lifetime_function(self) -> AsyncIterator[None]:
    async with get_session().create_client(
        "s3",
        **self.s3_client_kwargs,
        config=Config(signature_version="v4"),
    ) as self._client:  # type: ignore
        if not await s3_bucket_exists(self._client, self.bucket_name):
            if not self.auto_create_bucket:
                raise ValueError(
                    f"Bucket {self.bucket_name} does not exist and auto_create_bucket is disabled"
                )
            try:
                await self._client.create_bucket(Bucket=self.bucket_name)
            except ClientError as e:
                raise ValueError(
                    f"Failed to create bucket {self.bucket_name}"
                ) from e

        yield

Functions