app.core package


app.core.config module

class app.core.config.Settings(_case_sensitive: bool | None = None, _env_prefix: str | None = None, _env_file: DotenvType | None = PosixPath('.'), _env_file_encoding: str | None = None, _env_ignore_empty: bool | None = None, _env_nested_delimiter: str | None = None, _env_parse_none_str: str | None = None, _secrets_dir: str | Path | None = None, *, app_name: str = 'Flux RESTFul API', api_version: str = 'v1', has_gpus: bool = False, flux_nodes: int = 1, require_auth: bool = False, db_file: str = 'sqlite:///./flux-restful.db', flux_user: str = 'fluxuser', flux_token: str | None = None, flux_server_mode: str | None = 'single-user', secret_key: str = 'lTHpFxFnVV41sL8ZkwM9uR9yp0nAfygp', access_token_expires_minutes: int = 600, option_flags: dict = {}, known_launchers: list = ['nextflow', 'snakemake'])[source]

Bases: BaseSettings

Basic settings and defaults for the Flux RESTFul API

access_token_expires_minutes : int
api_version : str
app_name : str
db_file : str
flux_nodes : int
flux_server_mode : str | None
flux_token : str | None
flux_user : str
has_gpus : bool
known_launchers : list
model_computed_fields : ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config : ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_parse_none_str': None, 'env_prefix': '', 'extra': 'forbid', 'json_file': None, 'json_file_encoding': None, 'protected_namespaces': ('model_', 'settings_'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_file': None, 'yaml_file_encoding': None}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields : ClassVar[dict[str, FieldInfo]] = {'access_token_expires_minutes': FieldInfo(annotation=int, required=False, default=600), 'api_version': FieldInfo(annotation=str, required=False, default='v1'), 'app_name': FieldInfo(annotation=str, required=False, default='Flux RESTFul API'), 'db_file': FieldInfo(annotation=str, required=False, default='sqlite:///./flux-restful.db'), 'flux_nodes': FieldInfo(annotation=int, required=False, default=1), 'flux_server_mode': FieldInfo(annotation=Union[str, NoneType], required=False, default='single-user'), 'flux_token': FieldInfo(annotation=Union[str, NoneType], required=False), 'flux_user': FieldInfo(annotation=str, required=False, default='fluxuser'), 'has_gpus': FieldInfo(annotation=bool, required=False, default=False), 'known_launchers': FieldInfo(annotation=list, required=False, default=['nextflow', 'snakemake']), 'option_flags': FieldInfo(annotation=dict, required=False, default={}), 'require_auth': FieldInfo(annotation=bool, required=False, default=False), 'secret_key': FieldInfo(annotation=str, required=False, default='lTHpFxFnVV41sL8ZkwM9uR9yp0nAfygp')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

option_flags : dict
require_auth : bool
secret_key : str

Generate a secret key to encrypt, if one not provided.

app.core.config.get_bool_envar(key, default=False)[source]

Get a boolean from the environment, meaning the value is set.

app.core.config.get_int_envar(key, default=None)[source]

Get (and parse) an integer environment variable

app.core.config.get_option_flags(key, prefix='-o')[source]

Wrapper around parse_option_flags to get from environment.

The function can then be shared to parse flags from the UI in the same way.

app.core.config.parse_option_flags(flags, prefix='-o')[source]

Parse key value pairs (optionally with a prefix) from the environment.

app.core.logging module

app.core.logging.init_loggers(logger_name: str = 'errors-analysis')[source] module str | Any, expires_delta: timedelta | None = None, secret_key=None) str[source]

Create a jwt access token.

We either use the user’s secret key (which is hashed) or fall back to the server set secret key. str) str[source]

Note we aren’t providing a salt here, so the same password can generate different. str, hashed_password: str) bool[source]

Verify the password

Module contents

Last update: Mar 10, 2024