app.core package¶

Submodules¶

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¶

app.core.config.generate_secret_key(length=32)[source]¶

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]¶

app.core.security module¶

app.core.security.create_access_token(subject: 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.

app.core.security.get_password_hash(password: str) → str[source]¶

Note we aren’t providing a salt here, so the same password can generate different.

app.core.security.verify_password(plain_password: str, hashed_password: str) → bool[source]¶

Verify the password

Module contents¶