foca.models package

Submodules

foca.models.config module

FOCA config models.

class foca.models.config.APIConfig(*, specs: List[SpecConfig] = [])

Bases: FOCABaseConfig

Model for a list of configuration parameters for OpenAPI 2.x or 3.x specifications to be attached to a Connexion app.

Parameters:

specs – List of configuration parameters for OpenAPI 2.x or 3.x specifications to be attached to a Connexion app.

specs

List of configuration parameters for OpenAPI 2.x or 3.x specifications to be attached to a Connexion app.

Type:

List[foca.models.config.SpecConfig]

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> APIConfig(
...     specs=[SpecConfig(path='/path/to/specs.yaml')],
... )
APIConfig(specs=[SpecConfig(path=[PosixPath('/path/to/specs.yaml')], path_out=PosixPath('/path/to/specs.modified.yaml'), append=None, add_operation_fields=None, add_security_fields=None, disable_auth=False, connexion=None)])
specs: List[SpecConfig]
class foca.models.config.AccessControlConfig(*, api_specs: str | None = None, api_controllers: str | None = None, db_name: str | None = None, collection_name: str | None = None, model: str | None = None, owner_headers: set | None = None, user_headers: set | None = None)

Bases: FOCABaseConfig

Model for setting access control configuration. For exact behaviour cf. https://github.com/casbin/pycasbin.

Parameters:

  • api_specs – Path to API spec definition. If None, the use default specs.

  • api_controllers – Path to API spec controller. If None, the use default specs.

  • db_name – Access control database name. If None, the use default specs.

  • collection_name – Access control collection name. If None, the use default specs.

  • model – Path to access control model configuration file. If None, the use default specs.

  • owner_headers – Owner (Admin) specific header property requirements for casbin.

  • user_headers – User specific header property requirements for casbin.

api_specs

Path to API spec definition.

Type:

str | None

api_controllers

Path to API spec controller.

Type:

str | None

db_name

Access control database name.

Type:

str | None

collection_name

Access control collection name.

Type:

str | None

model

Path to access control model configuration file.

Type:

str | None

owner_headers

Owner (Admin) specific header property requirements for casbin.

Type:

set | None

user_headers

User specific header property requirements for casbin.

Type:

set | None

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> AccessControlConfig(
...     api_specs='/path/to/spec.yaml',
...     api_controllers='/path/to/spec_server.py',
...     db_name='access_control_db',
...     collection_name='access_control_collection',
...     model='/path/to/policy.conf',
...     owner_headers={'X-User', 'X-Group'},
...     user_headers={'X-User'}
... )
AccessControlConfig(api_specs='/path/to/access_control_spec.yaml',api_controllers='/path/to/access_control_spec_server.py', db_name='access_control_db', collection_name='access_control_collection', model='/path/to/policy.conf', owner_headers={'X-User', 'X-Group'}, user_headers={'X-User'})
api_controllers: str | None
api_specs: str | None
collection_name: str | None
db_name: str | None
model: str | None
owner_headers: set | None
user_headers: set | None
classmethod validate_model_path(v: Path | None)

Resolve path relative to caller’s current working directory if no absolute path provided for model or Set to default file path for model if path is not provided.

class foca.models.config.AuthConfig(*, required: bool = True, add_key_to_claims: bool = True, allow_expired: bool = False, audience: List[str] | None = None, claim_identity: str = 'sub', claim_issuer: str = 'iss', algorithms: List[str] = ['RS256'], validation_methods: List[ValidationMethodsEnum] = [ValidationMethodsEnum.userinfo, ValidationMethodsEnum.public_key], validation_checks: ValidationChecksEnum = ValidationChecksEnum.all)

Bases: FOCABaseConfig

Model for parameters used to configure JSON Web Token (JWT)-based authorization for the app.

Parameters:

  • required – Boolean to define the auth configuration for the app. Defaults to true.

  • add_key_to_claims – Whether to allow the application to add the identity provider’s corresponding JSON Web Key (JWK), in PEM format, to the dictionary of claims.

  • allow_expired – Allow/disallow expired JWTs. If False, a 401 authorization error is raised in response to a request containing an expired JWT.

  • audience – List of audiences that the app identifies itself with. If specified, JWTs that do not contain any of the specified audiences are rejected. If None, audience validation is disabled.

  • claim_identity – The JWT claim used to identify the sender.

  • claim_issuer – The JWT claim used to identify the issuer.

  • algorithms – Lists the JWT-signing algorithms supported by the app.

  • validation_methods – Lists the methods to be used to validate a JWT.

  • validation_checks – Specify how many of the validation_methods need to pass before accepting a JWT.

required

Boolean to define the auth configuration for the app.

Type:

bool

add_key_to_claims

Whether to allow the application to add the identity provider’s corresponding JSON Web Key (JWK), in PEM format, to the dictionary of claims.

Type:

bool

allow_expired

Allow/disallow expired JWTs. If False, a 401 authorization error is raised in response to a request containing an expired JWT.

Type:

bool

audience

List of audiences that the app identifies itself with. If specified, JWTs that do not contain any of the specified audiences are rejected. If None, audience validation is disabled.

Type:

List[str] | None

claim_identity

The JWT claim used to identify the sender.

Type:

str

claim_issuer

The JWT claim used to identify the issuer.

Type:

str

algorithms

Lists the JWT-signing algorithms supported by the app.

Type:

List[str]

validation_methods

Lists the methods to be used to validate a JWT.

Type:

List[foca.models.config.ValidationMethodsEnum]

validation_checks

Specify how many of the validation_methods need to pass before accepting a JWT.

Type:

foca.models.config.ValidationChecksEnum

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> AuthConfig(
...     required=False,
...     add_key_to_claims=True,
...     allow_expired=False,
...     audience=None,
...     claim_identity="sub",
...     claim_issuer="iss",
...     algorithms=["RS256"],
...     validation_methods=["userinfo", "public_key"],
...     validation_checks="all",
... )
AuthConfig(required=False, add_key_to_claims=True, allow_expired=False, audience=None, claim_identity='sub', claim_issuer='iss', algorithms=['RS256'], validation_methods=[<ValidationMethodsEnum.userinfo: 'userinfo'>, <ValidationMethodsEnum.public_key: 'public_key'>], validation_checks=<ValidationChecksEnum.all: 'all'>)
add_key_to_claims: bool
algorithms: List[str]
allow_expired: bool
audience: List[str] | None
claim_identity: str
claim_issuer: str
required: bool
validation_checks: ValidationChecksEnum
validation_methods: List[ValidationMethodsEnum]
class foca.models.config.CORSConfig(*, enabled: bool = True)

Bases: FOCABaseConfig

Model for Cross Origin Resource Sharing (CORS) configuration.

Parameters:

enabled – Enable/disable the CORS for the application.

enabled

Enable/disable the CORS for the application.

Type:

bool

Example

>>> CORSConfig(
...     enabled=True,
... )
CORSConfig(enabled=True)
enabled: bool
class foca.models.config.CollectionConfig(*, indexes: List[IndexConfig] | None = None, client: Collection | None = None)

Bases: FOCABaseConfig

Model for configuring a MongoDB collection.

Parameters:

  • indexes – An index configuration object.

  • client – Client connected to collection. Most likely populated through the code, not during setup.

indexes

An index configuration object.

Type:

List[foca.models.config.IndexConfig] | None

client

Client connected to collection. Most likely populated through the code, not during setup.

Type:

pymongo.collection.Collection | None

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> CollectionConfig(
...     indexes=[IndexConfig(keys={'last_name': 1})],
... )
CollectionConfig(indexes=[IndexConfig(keys=[('last_name', 1)], options={})], client=None)}, client=None)
client: Collection | None
indexes: List[IndexConfig] | None
class foca.models.config.Config(*, server: ~foca.models.config.ServerConfig = ServerConfig(host='0.0.0.0', port=8080, debug=True, environment='development', testing=False, use_reloader=True), exceptions: ~foca.models.config.ExceptionConfig = ExceptionConfig(required_members=[['title'], ['status']], extension_members=False, status_member=['status'], public_members=None, private_members=None, exceptions='foca.errors.exceptions.exceptions', logging=<ExceptionLoggingEnum.oneline: 'oneline'>, mapping={<class 'Exception'>: {'title': 'Internal Server Error', 'status': 500}, <class 'werkzeug.exceptions.BadRequest'>: {'title': 'Bad Request', 'status': 400}, <class 'connexion.exceptions.ExtraParameterProblem'>: {'title': 'Bad Request', 'status': 400}, <class 'werkzeug.exceptions.Unauthorized'>: {'title': 'Unauthorized', 'status': 401}, <class 'connexion.exceptions.OAuthProblem'>: {'title': 'Unauthorized', 'status': 401}, <class 'werkzeug.exceptions.Forbidden'>: {'title': 'Forbidden', 'status': 403}, <class 'werkzeug.exceptions.NotFound'>: {'title': 'Not Found', 'status': 404}, <class 'werkzeug.exceptions.InternalServerError'>: {'title': 'Internal Server Error', 'status': 500}, <class 'werkzeug.exceptions.BadGateway'>: {'title': 'Bad Gateway', 'status': 502}, <class 'werkzeug.exceptions.ServiceUnavailable'>: {'title': 'Service Unavailable', 'status': 502}, <class 'werkzeug.exceptions.GatewayTimeout'>: {'title': 'Gateway Timeout', 'status': 504}}), api: ~foca.models.config.APIConfig = APIConfig(specs=[]), security: ~foca.models.config.SecurityConfig = SecurityConfig(access_control=AccessControlConfig(api_specs=None, api_controllers=None, db_name=None, collection_name=None, model='/home/docs/checkouts/readthedocs.org/user_builds/foca/checkouts/latest/foca/security/access_control/api/default_model.conf', owner_headers=None, user_headers=None), auth=AuthConfig(required=True, add_key_to_claims=True, allow_expired=False, audience=None, claim_identity='sub', claim_issuer='iss', algorithms=['RS256'], validation_methods=[<ValidationMethodsEnum.userinfo: 'userinfo'>, <ValidationMethodsEnum.public_key: 'public_key'>], validation_checks=<ValidationChecksEnum.all: 'all'>), cors=CORSConfig(enabled=True)), db: ~foca.models.config.MongoConfig | None = None, jobs: ~foca.models.config.JobsConfig | None = None, log: ~foca.models.config.LogConfig = LogConfig(version=1, disable_existing_loggers=False, formatters={'standard': LogFormatterConfig(class_formatter='logging.Formatter', style='{', format='[{asctime}: {levelname:<8}] {message} [{name}]')}, handlers={'console': LogHandlerConfig(class_handler='logging.StreamHandler', level=20, formatter='standard', stream='ext://sys.stderr')}, root=LogRootConfig(level=10, handlers=['console'])), **extra_data: ~typing.Any)

Bases: FOCABaseConfig

Model for all app configuration parameters.

Parameters:

  • server – Server config parameters.

  • exceptions – Exception handling parameters.

  • api – OpenAPI specification config parameters.

  • security – Security config parameters.

  • db – Database config parameters.

  • jobs – Background job config parameters.

  • log – Logger config parameters.

server

Server config parameters.

Type:

foca.models.config.ServerConfig

exceptions

Exception handling parameters.

Type:

foca.models.config.ExceptionConfig

api

OpenAPI specification config parameters.

Type:

foca.models.config.APIConfig

security

Security config parameters.

Type:

foca.models.config.SecurityConfig

db

Database config parameters.

Type:

foca.models.config.MongoConfig | None

jobs

Background job config parameters.

Type:

foca.models.config.JobsConfig | None

log

Logger config parameters.

Type:

foca.models.config.LogConfig

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> Config()
Config(server=ServerConfig(host='0.0.0.0', port=8080, debug=True, environment='development', testing=False, use_reloader=True), exceptions=ExceptionConfig(required_members=[['title'], ['status']], extension_members=False, status_member=['status'], public_members=None, private_members=None, exceptions='foca.errors.exceptions.exceptions', logging=<ExceptionLoggingEnum.oneline: 'oneline'>, mapping={<class 'Exception'>: {'title': 'Internal Server Error', 'status': 500}, <class 'werkzeug.exceptions.BadRequest'>: {'title': 'Bad Request', 'status': 400}, <class 'connexion.exceptions.ExtraParameterProblem'>: {'title': 'Bad Request', 'status': 400}, <class 'werkzeug.exceptions.Unauthorized'>: {'title': 'Unauthorized', 'status': 401}, <class 'connexion.exceptions.OAuthProblem'>: {'title': 'Unauthorized', 'status': 401}, <class 'werkzeug.exceptions.Forbidden'>: {'title': 'Forbidden', 'status': 403}, <class 'werkzeug.exceptions.NotFound'>: {'title': 'Not Found', 'status': 404}, <class 'werkzeug.exceptions.InternalServerError'>: {'title': 'Internal Server Error', 'status': 500}, <class 'werkzeug.exceptions.BadGateway'>: {'title': 'Bad Gateway', 'status': 502}, <class 'werkzeug.exceptions.ServiceUnavailable'>: {'title': 'Service Unavailable', 'status': 502}, <class 'werkzeug.exceptions.GatewayTimeout'>: {'title': 'Gateway Timeout', 'status': 504}}), api=APIConfig(specs=[]), security=SecurityConfig(auth=AuthConfig(required=False, add_key_to_claims=True, allow_expired=False, audience=None, claim_identity='sub', claim_issuer='iss', algorithms=['RS256'], validation_methods=[<ValidationMethodsEnum.userinfo: 'userinfo'>, <ValidationMethodsEnum.public_key: 'public_key'>], validation_checks=<ValidationChecksEnum.all: 'all'>), cors=CORSConfig(enabled=True), access_control=AccessControlConfig(api_specs='/path/to/access_control_spec.yaml', api_controllers='/path/to/access_control_spec_server.py', db_name='access_control_db', collection_name='access_control_collection', model='/path/to/policy.conf', owner_headers={'X-User', 'X-Group'}, user_headers={'X-User'})), db=None, jobs=None, log=LogConfig(version=1, disable_existing_loggers=False, formatters={'standard': LogFormatterConfig(class_formatter='logging.Formatter', style='{', format='[{asctime}: {levelname:<8}] {message} [{name}]')}, handlers={'console': LogHandlerConfig(class_handler='logging.StreamHandler', level=20, formatter='standard', stream='ext://sys.stderr')}, root=LogRootConfig(level=10, handlers=['console'])))
class Config

Bases: object

Configuration for Pydantic model class.

extra = 'allow'
api: APIConfig
db: MongoConfig | None
exceptions: ExceptionConfig
jobs: JobsConfig | None
log: LogConfig
security: SecurityConfig
server: ServerConfig
class foca.models.config.DBConfig(*, collections: Dict[str, CollectionConfig] | None = None, client: Database | None = None)

Bases: FOCABaseConfig

Model for configuring a MongoDB database.

Parameters:

  • collections – Mapping of collection names (keys) and configuration objects (values).

  • client – Client connected to database. Most likely populated through the code, not during setup.

collections

Mapping of collection names (keys) and configuration objects (values).

Type:

Dict[str, foca.models.config.CollectionConfig] | None

client

Client connected to database. Most likely populated through the code, not during setup.

Type:

pymongo.database.Database | None

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> DBConfig(
...     collections={
...         'my_collection': CollectionConfig(
...             indexes=[IndexConfig(keys={'last_name': 1})],
...         ),
...     },
... )
DBConfig(collections={'my_collection': CollectionConfig(indexes=[IndexConfig(keys=[('last_name', 1)], options={})], client=None)}, client=None)
client: Database | None
collections: Dict[str, CollectionConfig] | None
class foca.models.config.ExceptionConfig(*, required_members: List[List[str]] = [['title'], ['status']], extension_members: bool | List[List[str]] = False, status_member: List[str] = ['status'], public_members: List[List[str]] | None = None, private_members: List[List[str]] | None = None, exceptions: str = 'foca.errors.exceptions.exceptions', logging: ExceptionLoggingEnum = ExceptionLoggingEnum.oneline, mapping: Dict[str, Dict[str, Any]] | None = None)

Bases: FOCABaseConfig

Model for app context JSON exceptions to be registered with a Connexion app.

Parameters:

  • required_members – List of dictionary keys indicating which JSON members are required for all exceptions. Must contain a member that represents the HTTP response code (cf. status_member).

  • extension_members – Either a list of additionally allowed, optional extension members, or a Boolean expression indicating whether any (True) or no (False) additional members are allowed.

  • status_member – Sequence of dictionary keys indicating the member that represents the HTTP response code (e.g, 500).

  • public_members – Filter to restrict which exception members are to be included in the error response. Only members listed here, with each one specified as a sequence of keys, are included. Specify an empty list to prevent any members from being returned to the user. Set to None to disable filtering. Note that only one of public_members and private_members filters can be active.

  • private_members – Filter to restrict which exception members are to be included in the error response. Members listed here, with each one specified as a sequence of keys, are excluded. Set to None to disable filtering. Note that only one of public_members and private_members filters can be active.

  • exceptions – Path to dictionary containing the actual exception classes as keys and a dictionary of JSON members (as per required_members and extension_members) as values. Path should be a dot-separated path to the module containing the dictionary (which needs to also contain imports for all listed exceptions), followed by the name of the dictionary itself. For example, for myapp.errors.exc_dict, the dictionary exc_dict would be attempted to be imported from module myapp.errors (must be available in the Pythonpath). To ensure that all exceptions arising during the app context are handled, it is strongly advised to add the catch-all exception Exception. If missing, any exceptions not listed will only provoke an empty JSON response.

  • logging – Specifies if and how exception details should be logged. Note that unless foca.models.config.ExceptionLoggingEnum.none is specified, a JSON representation of the error, as defined in exceptions, and including _all_ members, unaffected by public_members and private_members filters, will be logged on an additional line.

  • mapping – The actual referenced dictionary from exceptions, populated by FOCA.

required_members

List of dictionary keys indicating which JSON members are required for all exceptions. Must contain a member that represents the HTTP response code (cf. status_member).

Type:

List[List[str]]

extension_members

Either a list of additionally allowed, optional extension members, or a Boolean expression indicating whether any (True) or no (False) additional members are allowed.

Type:

bool | List[List[str]]

status_member

Sequence of dictionary keys indicating the member that represents the HTTP response code (e.g, 500).

Type:

List[str]

public_members

Filter to restrict which exception members are to be included in the error response. Only members listed here, with each one specified as a sequence of keys, are included. Specify an empty list to prevent any members from being returned to the user. Set to None to disable filtering. Note that only one of public_members and private_members filters can be active.

Type:

List[List[str]] | None

private_members

Filter to restrict which exception members are to be included in the error response. Members listed here, with each one specified as a sequence of keys, are excluded. Set to None to disable filtering. Note that only one of public_members and private_members filters can be active.

Type:

List[List[str]] | None

exceptions

Path to dictionary containing the actual exception classes as keys and a dictionary of JSON members (as per required_members and extension_members) as values. Path should be a dot-separated path to the module containing the dictionary (which needs to also contain imports for all listed exceptions), followed by the name of the dictionary itself. For example, for myapp.errors.exc_dict, the dictionary exc_dict would be attempted to be imported from module myapp.errors (must be available in the Pythonpath). To ensure that all exceptions arising during the app context are handled, it is strongly advised to add the catch-all exception Exception. If missing, any exceptions not listed will only provoke an empty JSON response.

Type:

str

logging

Specifies if and how exception details should be logged. Note that unless foca.models.config.ExceptionLoggingEnum.none is specified, a JSON representation of the error, as defined in exceptions, and including _all_ members, unaffected by public_members and private_members filters, will be logged on an additional line.

Type:

foca.models.config.ExceptionLoggingEnum

mapping

The actual referenced dictionary from exceptions, populated by FOCA.

Type:

Dict[str, Dict[str, Any]] | None

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> ExceptionConfig()
ExceptionConfig(required_members=[['title'], ['status']], extension_members=False, status_member=['status'], public_members=None, private_members=None, exceptions='foca.errors.exceptions.exceptions', logging=<ExceptionLoggingEnum.oneline: 'oneline'>, mapping={<class 'Exception'>: {'title': 'Internal Server Error', 'status': 500}, <class 'werkzeug.exceptions.BadRequest'>: {'title': 'Bad Request', 'status': 400}, <class 'connexion.exceptions.ExtraParameterProblem'>: {'title': 'Bad Request', 'status': 400}, <class 'werkzeug.exceptions.Unauthorized'>: {'title': 'Unauthorized', 'status': 401}, <class 'connexion.exceptions.OAuthProblem'>: {'title': 'Unauthorized', 'status': 401}, <class 'werkzeug.exceptions.Forbidden'>: {'title': 'Forbidden', 'status': 403}, <class 'werkzeug.exceptions.NotFound'>: {'title': 'Not Found', 'status': 404}, <class 'werkzeug.exceptions.InternalServerError'>: {'title': 'Internal Server Error', 'status': 500}, <class 'werkzeug.exceptions.BadGateway'>: {'title': 'Bad Gateway', 'status': 502}, <class 'werkzeug.exceptions.ServiceUnavailable'>: {'title': 'Service Unavailable', 'status': 502}, <class 'werkzeug.exceptions.GatewayTimeout'>: {'title': 'Gateway Timeout', 'status': 504}})
exceptions: str
extension_members: bool | List[List[str]]
logging: ExceptionLoggingEnum
mapping: Dict[str, Dict[str, Any]] | None
private_members: List[List[str]] | None
public_members: List[List[str]] | None
required_members: List[List[str]]
status_member: List[str]
classmethod validate_mapping(v, *, values)

Validate that exceptions dictionary exists and can be imported, that all exceptions have all required members and no additional members (unless specifically allowed) and replace default value for field mapping to the contents of the exceptions dictionary.

class foca.models.config.ExceptionLoggingEnum(value)

Bases: Enum

Enumerator for exception logging config values.

minimal

Exception title and message are logged on a single line.

none

Exception details are not logged.

oneline

Exception, including traceback, is logged on a single line.

regular

The exception is logged with the entire traceback stack, typically on multiple lines.

minimal = 'minimal'
none = 'none'
oneline = 'oneline'
regular = 'regular'
class foca.models.config.FOCABaseConfig

Bases: BaseModel

Base configuration for FOCA models.

class Config

Bases: object

Configuration for Pydantic model class.

arbitrary_types_allowed = True
extra = 'forbid'
class foca.models.config.IndexConfig(*, keys: Dict | None = None, options: Dict = {})

Bases: FOCABaseConfig

Model for configuring indexes for a MongoDB collection.

Parameters:
keys

A key-direction dictionary indicating the field to be indexed and the sort order of that index.

Type:

Dict | None

options

A dictionary of any additional index creation options. cf.: https://pymongo.readthedocs.io/en/3.10.1/api/pymongo/collection.html

Type:

Dict

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> IndexConfig(
...     keys={'name': -1, 'id': 1},
...     options={'unique': True, 'sparse': False}
... )
IndexConfig(keys=[('name', -1), ('id', 1)], options={'unique': True, 'sparse': False})
keys: Dict | None
options: Dict
classmethod store_enum_value(v)

Convert dict values of keys into list of tuples

class foca.models.config.JobsConfig(*, host: str = 'rabbitmq', port: int = 5672, backend: str = 'rpc://', include: List[str] | None = None)

Bases: FOCABaseConfig

Model for configuring a RabbitMQ broker instance and Celery client and tasks to be attached to a Flask/Connexion app.

Parameters:

  • host – Host at which the broker is exposed.

  • port – Port at which the broker is exposed.

  • backend – Backend used to store background task results.

  • include – List of modules to import when workers start.

host

Host at which the broker is exposed.

Type:

str

port

Port at which the broker is exposed.

Type:

int

backend

Backend used to store background task results.

Type:

str

include

List of modules to import when workers start.

Type:

List[str] | None

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> JobsConfig(
...     host="rabbitmq",
...     port=5672,
...     backend='rpc://',
...     include=[],
... )
JobsConfig(host='rabbitmq', port=5672, backend='rpc://', include=[])
backend: str
host: str
include: List[str] | None
port: int
class foca.models.config.LogConfig(*, version: int = 1, disable_existing_loggers: bool = False, formatters: Dict[str, LogFormatterConfig] | None = {'standard': LogFormatterConfig(class_formatter='logging.Formatter', style='{', format='[{asctime}: {levelname:<8}] {message} [{name}]')}, handlers: Dict[str, LogHandlerConfig] | None = {'console': LogHandlerConfig(class_handler='logging.StreamHandler', level=20, formatter='standard', stream='ext://sys.stderr')}, root: LogRootConfig | None = LogRootConfig(level=10, handlers=['console']))

Bases: FOCABaseConfig

Model for passing parameters for configuring app logging.

Parameters:

  • version – Schema version.

  • disable_existing_loggers – Whether any existing non-root loggers are to be disabled.

  • formatters – A dictionary of logging formatters, where keys represent formatter names and values represent the corresponding configuration parameters.

  • handlers – A dictionary of logging handlers, where keys represent handler names and values represent the corresponding configuration parameters.

  • root – Configuration of the root logger.

version

Represents current schema version.

Type:

int

disable_existing_loggers

Whether any existing non-root loggers are to be disabled.

Type:

bool

formatters

A dictionary of logging formatters, where keys represent formatter names and values represent the corresponding configuration parameters.

Type:

Dict[str, foca.models.config.LogFormatterConfig] | None

handlers

A dictionary of logging handlers, where keys represent handler names and values represent the corresponding configuration parameters.

Type:

Dict[str, foca.models.config.LogHandlerConfig] | None

root

Configuration of the root logger.

Type:

foca.models.config.LogRootConfig | None

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> LogConfig(
...     version=1,
...     disable_existing_loggers=False,
...     formatters={
...         "standard": LogFormatterConfig()
...     },
...     handlers={
...         "console": LogHandlerConfig()
...     },
...     root=LogRootConfig()
... )
LogConfig(version=1, disable_existing_loggers=False, formatters={'standard': LogFormatterConfig(class_formatter='logging.Formatter', style='{', format='[{asctime}: {levelname:<8}] {message} [{name}]')}, handlers={'console': LogHandlerConfig(class_handler='logging.StreamHandler', level=20, formatter='standard', stream='ext://sys.stderr')}, root=LogRootConfig(level=10, handlers=['console']))
disable_existing_loggers: bool
formatters: Dict[str, LogFormatterConfig] | None
handlers: Dict[str, LogHandlerConfig] | None
root: LogRootConfig | None
version: int
class foca.models.config.LogFormatterConfig(*, style: str = '{', format: str = '[{asctime}: {levelname:<8}] {message} [{name}]', **extra_data: Any)

Bases: FOCABaseConfig

Model for formatter for LogConfig.

Parameters:

  • class – Name of logging formatter class.

  • style – Determines how the format string will be merged with its data.

  • format – Format string any log messages.

class

Name of logging formatter class.

style

Determines how the format string will be merged with its data.

Type:

str

format

Format string any log messages.

Type:

str

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> LogFormatterConfig(
...     style="{",
...     format="[{asctime}: {levelname:<8}] {message} [{name}]",
... )
LogFormatterConfig(class_formatter='logging.Formatter', style='{', format='[{asctime}: {levelname:<8}] {message} [{name}]')
class_formatter: str
format: str
style: str
class foca.models.config.LogHandlerConfig(*, level: int = 20, formatter: str = 'standard', stream: str = 'ext://sys.stderr', **extra_data: Any)

Bases: FOCABaseConfig

Model for passing logging handler parameters to configuring app logging.

Parameters:

  • class – Name of logging handler class.

  • level – Numeric value of logging level.

  • formatter – Name of logging formatter.

  • stream – Devide to which log is streamed.

class

Name of logging handler class.

level

Numeric value of logging level.

Type:

int

formatter

Name of logging formatter.

Type:

str

stream

Devide to which log is streamed.

Type:

str

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> LogHandlerConfig(
...     level=20,
...     formatter="standard",
...     stream="ext://sys.stderr",
... )
LogHandlerConfig(class_handler='logging.StreamHandler', level=20, formatter='standard', stream='ext://sys.stderr')
class_handler: str
formatter: str
level: int
stream: str
class foca.models.config.LogRootConfig(*, level: int = 10, handlers: List[str] | None = ['console'])

Bases: FOCABaseConfig

Model for root log configuration.

Parameters:

  • level – Numeric value of logging level.

  • handlers – List of logging handlers by name.

level

Numeric value of logging level.

Type:

int

handlers

List of logging handlers by name.

Type:

List[str] | None

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> LogRootConfig(
...     level=logging.INFO,
...     handlers=["console"],
... )
LogRootConfig(level=20, handlers=['console'])
handlers: List[str] | None
level: int
class foca.models.config.MongoConfig(*, host: str = 'mongodb', port: int = 27017, dbs: Dict[str, DBConfig] | None = None)

Bases: FOCABaseConfig

Model for configuring a MongoDB instance attached to a Flask or Connexion app.

Parameters:

  • host – Host at which the database is exposed.

  • port – Port at which the database is exposed.

  • dbs – Mapping of database names (keys) and configuration objects (values).

host

Host at which the database is exposed.

Type:

str

port

Port at which the database is exposed.

Type:

int

dbs

Mapping of database names (keys) and configuration objects (values).

Type:

Dict[str, foca.models.config.DBConfig] | None

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> MongoConfig(
...     host="mongodb",
...     port=27017,
... )
MongoConfig(host='mongodb', port=27017, dbs=None)
dbs: Dict[str, DBConfig] | None
host: str
port: int
class foca.models.config.PymongoDirectionEnum(value)

Bases: Enum

Enumerator for supported Pymongo index directions.

Cf. https://pymongo.readthedocs.io/en/3.10.1/api/pymongo/collection.html#pymongo.collection.Collection.create_index

ASCENDING

Ascending sort order.

DESCENDING

Descending sort order.

GEO2D

Index specifier for a 2-dimensional geospatial index.

GEOHAYSTACK

Index specifier for a 2-dimensional haystack index.

GEOSPHERE

Index specifier for a spherical geospatial index.

HASHED

Index specifier for a hashed index.

TEXT

Index specifier for a text index.

ASCENDING = 1
DESCENDING = -1
GEO2D = '2d'
GEOHAYSTACK = 'geoHaystack'
GEOSPHERE = '2dsphere'
HASHED = 'hashed'
TEXT = 'text'
class foca.models.config.SecurityConfig(*, access_control: ~foca.models.config.AccessControlConfig = AccessControlConfig(api_specs=None, api_controllers=None, db_name=None, collection_name=None, model='/home/docs/checkouts/readthedocs.org/user_builds/foca/checkouts/latest/foca/security/access_control/api/default_model.conf', owner_headers=None, user_headers=None), auth: ~foca.models.config.AuthConfig = AuthConfig(required=True, add_key_to_claims=True, allow_expired=False, audience=None, claim_identity='sub', claim_issuer='iss', algorithms=['RS256'], validation_methods=[<ValidationMethodsEnum.userinfo: 'userinfo'>, <ValidationMethodsEnum.public_key: 'public_key'>], validation_checks=<ValidationChecksEnum.all: 'all'>), cors: ~foca.models.config.CORSConfig = CORSConfig(enabled=True))

Bases: FOCABaseConfig

Model for list the Security configuration.

Parameters:

  • access_control – Config parameters for Access Control.

  • auth – Config parameters for JSON Web Token (JWT) validation.

  • cors – Config parameters for CORS.

access_control

Config parameters for Access Control.

Type:

foca.models.config.AccessControlConfig

auth

Config parameters for JSON Web Token (JWT) validation.

Type:

foca.models.config.AuthConfig

cors

Config parameters for CORS.

Type:

foca.models.config.CORSConfig

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> SecurityConfig(
...     access_control=AccessControlConfig(),
...     auth=AuthConfig(),
...     cors=CORSConfig(),
... )
SecurityConfig(auth=AuthConfig(required=False, add_key_to_claims=True, allow_expired=False, audience=None, claim_identity='sub', claim_issuer='iss', algorithms=['RS256'], validation_methods=[<ValidationMethodsEnum.userinfo: 'userinfo'>, <ValidationMethodsEnum.public_key: 'public_key'>], validation_checks=<ValidationChecksEnum.all: 'all'>), cors=CORSConfig(enabled=True), access_control=AccessControlConfig(api_specs='/path/to/access_control_spec.yaml',api_controllers='/path/to/access_control_spec_server.py', db_name='access_control_db', collection_name='access_control_collection', model='/path/to/policy.conf', owner_headers={'X-User', 'X-Group'}, user_headers={'X-User'}))
access_control: AccessControlConfig
auth: AuthConfig
cors: CORSConfig
class foca.models.config.ServerConfig(*, host: str = '0.0.0.0', port: int = 8080, debug: bool = True, environment: str = 'development', testing: bool = False, use_reloader: bool = True)

Bases: FOCABaseConfig

Model for configuration parameters to set up a Flask or Connexion app instance.

Parameters:

  • host – Host at which the application is exposed.

  • port – Port at which the application is exposed.

  • debug – Flag to run application in debug mode. If True, the application runs in debug mode and an interactive debugger will be shown for unhandled exceptions. See Flask documentation for more details.

  • environment – Variable to specify the application environment variable. See Flask documentation for more details.

  • testing – Enable/disable testing mode. If True, exceptions are propagated rather than handled by the the app’s error handlers.

  • use_reloader – Enable/disable the application reloader. If debug=True, enabling this will allow the server to reload automatically on code changes. See Flask documentation for more details.

host

Host at which the application is exposed.

Type:

str

port

Port at which the application is exposed.

Type:

int

debug

Flag to run application in debug mode. If True, the application runs in debug mode and an interactive debugger will be shown for unhandled exceptions. See Flask documentation for more details.

Type:

bool

environment

Variable to specify the application environment variable. See Flask documentation for more details.

Type:

str

testing

Enable/disable testing mode. If True, exceptions are propagated rather than handled by the the app’s error handlers.

Type:

bool

use_reloader

Enable/disable the application reloader. If debug=True, enabling this will allow the server to reload automatically on code changes. See Flask documentation for more details.

Type:

bool

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> ServerConfig(
...     host="0.0.0.0",
...     port=8080,
...     debug=True,
...     environment="development",
...     testing=False,
...     use_reloader=True,
... )
ServerConfig(host='0.0.0.0', port=8080, debug=True, environment='development', testing=False, use_reloader=True)
debug: bool
environment: str
host: str
port: int
testing: bool
use_reloader: bool
class foca.models.config.SpecConfig(*, path: Path | List[Path], path_out: Path | None = None, append: List[Dict] | None = None, add_operation_fields: Dict | None = None, add_security_fields: Dict | None = None, disable_auth: bool = False, connexion: Dict | None = None)

Bases: FOCABaseConfig

Model for configuration parameters for OpenAPI 2.x or 3.x specifications to be attached to a Connexion app.

Parameters:
path

A single path or list of paths to OpenAPI 2.x or 3.x specification in YAML format.

Type:

pathlib.Path | List[pathlib.Path]

path_out

Output path for modified specification file. Ignored if specs are not modified. If not specified, the original file path is stripped of the file extension and the suffix .modified.yaml is appended.

Type:

pathlib.Path | None

append

Fields to be added/modified in the root of the specification file. For OpenAPI 2.x, see https://swagger.io/specification/v2/. For OpenAPI 3.x, see https://swagger.io/specification/.

Type:

List[Dict] | None

add_operation_fields

Fields to be added/modified to/in the Operation Objects of each Path Info Object. An example use case for this is the addition or replacement of the x-swagger-router-controller field. For OpenAPI 2.x, see https://swagger.io/specification/v2/#operation-object. For OpenAPI 3.x, see https://swagger.io/specification/#operation-object. Note that different operation fields for different Operation Objects are currently not supported.

Type:

Dict | None

add_security_fields

Fields to be added/modified to/in each definition or scheme in the securityDefintions (OpenAPI 2.x) or securitySchemes (OpenAPI 3.x) objects. An example use case for this is the addition or replacement of the x-tokenInfoFunc or similar field. Cf. https://connexion.readthedocs.io/en/latest/security.html. For OpenAPI 2.x, see https://swagger.io/specification/v2/#securityDefinitionsObject. For OpenAPI 3.x, see https://swagger.io/docs/specification/authentication/. Note that different security fields for different security definitions/schemes are currently not supported.

Type:

Dict | None

disable_auth

Disable JWT validation for endpoints configured to require authorization as per the OpenAPI specifications. Has no effect if relevant security definitions/schemes are not defined. Setting is global. Use the security property in the OpenAPI specification to define this behavior separately for each Operation Object and/or security definition/scheme. For OpenAPI 2.x, see https://swagger.io/specification/v2/#securityDefinitionsObject. For OpenAPI 3.x, see https://swagger.io/docs/specification/authentication/.

Type:

bool

connexion

Keyword arguments passed through to connexion.apps.flask_app.add_api().

Type:

Dict | None

Raises:

pydantic.ValidationError – The class was instantianted with an illegal data type.

Example

>>> SpecConfig(path="/my/path.yaml")
SpecConfig(path=[PosixPath('/my/path.yaml')], path_out=PosixPath('/my/path.modified.yaml'), append=None, add_operation_fields=None, add_security_fields=None, disable_auth=False, connexion=None)

>>> SpecConfig(
...     path=["/path/to/specs.yaml", "/path/to/add_specs.yaml"],
...     path_out="/path/to/specs.modified.yaml",
...     append=[
...         {
...             "security": {
...                 "jwt": {
...                     "type": "apiKey",
...                     "name": "Authorization",
...                     "in": "header",
...                 }
...             }
...         },
...         {
...             "my_other_root_field": "some_value",
...         },
...     ],
...     add_operation_fields = {
...         "x-swagger-router-controller": "controllers.my_specs",
...         "x-some-other-custom-field": "some_value",
...     },
...     add_security_fields = {
...         "x-apikeyInfoFunc": "security.auth.validate_token",
...         "x-some-other-custom-field": "some_value",
...     },
...     disable_auth = False
... )
SpecConfig(path=[PosixPath('/path/to/specs.yaml'), PosixPath('/path/to/add_specs.yaml')], path_out=PosixPath('/path/to/specs.modified.yaml'), append=[{'security': {'jwt': {'type': 'apiKey', 'name': 'Authorization', 'in': 'header'}}}, {'my_other_root_field': 'some_value'}], add_operation_fields={'x-swagger-router-controller': 'controllers.my_specs', 'x-some-other-custom-field': 'some_value'}, add_security_fields={'x-apikeyInfoFunc': 'security.auth.validate_token', 'x-some-other-custom-field': 'some_value'}, disable_auth=False, connexion=None)
add_operation_fields: Dict | None
add_security_fields: Dict | None
append: List[Dict] | None
connexion: Dict | None
disable_auth: bool
path: Path | List[Path]
path_out: Path | None
classmethod set_abs_path(v)

Resolve path relative to caller’s current working directory if no absolute path provided.

classmethod set_default_out_path(v, *, values)

Set default output path for spec file if not supplied by user.

class foca.models.config.ValidationChecksEnum(value)

Bases: Enum

Enumerator for values indicating how many JSON Web Token (JWT) validation methods are requested.

all

All JWT validation methods need to pass; validation fails after first unsuccessful check.

any

Any method is sufficient to validate the JWT; validation succeeds after the first successful check.

all = 'all'
any = 'any'
class foca.models.config.ValidationMethodsEnum(value)

Bases: Enum

Enumerator for JSON Web Token (JWT) validation methods.

public_key

JWT validation via the identity provider’s JSON Web Key.

userinfo

JWT validation via OpenID Connect-compliant identity provider’s /userinfo endpoint.

public_key = 'public_key'
userinfo = 'userinfo'