academy.exchange.cloud.config

academy/exchange/cloud/config.py

Cloud exchange configuration file parsing.

LogConfig

Bases: BaseModel

Log config for exchange server.

Attributes:

• level (int | str) –

  Console log verbosity.

• logfile (str | None) –

  Base file name to write to.

• logfile_level (int | str) –

  Verbosity of logs in file.

• rotate (bool) –

  Weather to rotate logs or not.

• rotation_interval_days (int) –

  How often in days to rotate logs.

• backup_count (int) –

  How many previous log files to keep

init_logger

init_logger() -> Logger

Initialize logger from this configuration.

Source code in academy/exchange/cloud/config.py

def init_logger(self) -> logging.Logger:
    """Initialize logger from this configuration."""
    stdout_handler = logging.StreamHandler(sys.stdout)
    stdout_handler.setFormatter(_Formatter())
    stdout_handler.setLevel(self.level)
    handlers: list[logging.Handler] = [stdout_handler]

    if self.logfile is not None:
        logfile_level = (
            self.level
            if self.logfile_level is None
            else self.logfile_level
        )
        path = pathlib.Path(self.logfile)
        path.parent.mkdir(parents=True, exist_ok=True)
        file_handler: logging.Handler
        if self.rotate:
            file_handler = TimedRotatingFileHandler(
                path,
                when='d',
                interval=self.rotation_interval_days,
                backupCount=self.backup_count,
            )
        else:
            file_handler = logging.FileHandler(path)

        file_handler.setFormatter(_Formatter(color=False))
        file_handler.setLevel(logfile_level)
        handlers.append(file_handler)

    logging.basicConfig(
        datefmt='%Y-%m-%d %H:%M:%S',
        level=logging.NOTSET,
        handlers=handlers,
    )

    # This needs to be after the configuration of the root logger because
    # warnings get logged to a 'py.warnings' logger.
    logging.captureWarnings(True)

    logger = logging.getLogger()
    logger.info(
        (
            'Configured logger (stdout-level=%s, '
            'logfile=%s, logfile-level=%s)'
        ),
        logging.getLevelName(self.level)
        if isinstance(self.level, int)
        else self.level,
        self.logfile,
        logging.getLevelName(self.logfile_level)
        if isinstance(self.logfile_level, int)
        else self.logfile_level,
    )
    return logger

ExchangeAuthConfig

Bases: BaseModel

Exchange authentication configuration.

Attributes:

• method (Literal['globus'] | None) –

  Authentication method.

• kwargs (dict[str, Any]) –

  Arbitrary keyword arguments to pass to the authenticator. The kwargs are excluded from the repr() of this class because they often contain secrets.

BackendConfig

Bases: Protocol

Config for backend of storing messages.

get_backend abstractmethod

get_backend() -> MailboxBackend

Construct an instance of the backend from the config.

Source code in academy/exchange/cloud/config.py

@abc.abstractmethod
def get_backend(self) -> MailboxBackend:
    """Construct an instance of the backend from the config."""
    ...

PythonBackendConfig

Bases: BaseModel

Config for using PythonBackend.

get_backend

get_backend() -> MailboxBackend

Construct an instance of the backend from the config.

Source code in academy/exchange/cloud/config.py

def get_backend(self) -> MailboxBackend:
    """Construct an instance of the backend from the config."""
    return PythonBackend()

RedisBackendConfig

Bases: BaseModel

Config for RedisBackend.

Attributes:

• hostname (str) –

  Redis host

• port (int) –

  Redis port

• kwargs (dict[str, Any]) –

  Any additional args to Redis

get_backend

get_backend() -> MailboxBackend

Construct an instance of the backend from the config.

Source code in academy/exchange/cloud/config.py

def get_backend(self) -> MailboxBackend:
    """Construct an instance of the backend from the config."""
    return RedisBackend(
        self.hostname,
        self.port,
        message_size_limit_kb=self.message_size_limit_kb,
        kwargs=self.kwargs,
        mailbox_expiration_s=int(self.mailbox_expiration_d * 24 * 3600),
        gravestone_expiration_s=int(
            self.gravestone_expiration_d * 24 * 3600,
        ),
    )

ExchangeServingConfig

Bases: BaseModel

Exchange serving configuration.

Attributes:

• host (str) –

  Network interface the server binds to.

• port (int) –

  Network port the server binds to.

• certfile (str | None) –

  Certificate file (PEM format) use to enable TLS.

• keyfile (str | None) –

  Private key file. If not specified, the key will be taken from the certfile.

• auth (ExchangeAuthConfig) –

  Authentication configuration.

• log_file (ExchangeAuthConfig) –

  Location to write logs.

• log_level (ExchangeAuthConfig) –

  Verbosity of logs.

from_toml classmethod

from_toml(filepath: str | Path) -> Self

Parse an TOML config file.

Example

Minimal config without SSL and without authentication.

exchange.toml

port = 8700

from academy_exchange.config import ExchangeServingConfig

config = ExchangeServingConfig.from_toml('exchange.toml')

Example

Serve with SSL and Globus Auth.

relay.toml

host = "0.0.0.0"
port = 8700
certfile = "/path/to/cert.pem"
keyfile = "/path/to/privkey.pem"

[auth]
method = "globus"

[auth.kwargs]
client_id = "..."
client_secret = "..."

Note

Omitted values will be set to their defaults (if they are an optional value with a default).

relay.toml

[serving]
certfile = "/path/to/cert.pem"

from academy_exchange.config import ExchangeServingConfig

config = ExchangeServingConfig.from_config('relay.toml')
assert config.certfile == '/path/to/cert.pem'
assert config.keyfile is None

Source code in academy/exchange/cloud/config.py

@classmethod
def from_toml(cls, filepath: str | pathlib.Path) -> Self:
    """Parse an TOML config file.

    Example:
        Minimal config without SSL and without authentication.
        ```toml title="exchange.toml"
        port = 8700
        ```

        ```python
        from academy_exchange.config import ExchangeServingConfig

        config = ExchangeServingConfig.from_toml('exchange.toml')
        ```

    Example:
        Serve with SSL and Globus Auth.
        ```toml title="relay.toml"
        host = "0.0.0.0"
        port = 8700
        certfile = "/path/to/cert.pem"
        keyfile = "/path/to/privkey.pem"

        [auth]
        method = "globus"

        [auth.kwargs]
        client_id = "..."
        client_secret = "..."
        ```

    Note:
        Omitted values will be set to their defaults (if they are an
        optional value with a default).
        ```toml title="relay.toml"
        [serving]
        certfile = "/path/to/cert.pem"
        ```

        ```python
        from academy_exchange.config import ExchangeServingConfig

        config = ExchangeServingConfig.from_config('relay.toml')
        assert config.certfile == '/path/to/cert.pem'
        assert config.keyfile is None
        ```
    """
    with open(filepath, 'rb') as f:
        return load(cls, f)

load

load(model: type[BaseModelT], fp: BinaryIO) -> BaseModelT

Parse TOML from a binary file to a data class.

Parameters:

• model (type[BaseModelT]) –

  Config model type to parse TOML using.

• fp (BinaryIO) –

  File-like bytes stream to read in.

Returns:

• BaseModelT –

  Model initialized from TOML file.

Source code in academy/exchange/cloud/config.py

def load(model: type[BaseModelT], fp: BinaryIO) -> BaseModelT:
    """Parse TOML from a binary file to a data class.

    Args:
        model: Config model type to parse TOML using.
        fp: File-like bytes stream to read in.

    Returns:
        Model initialized from TOML file.
    """
    return loads(model, fp.read().decode())

loads

loads(model: type[BaseModelT], data: str) -> BaseModelT

Parse TOML string to data class.

Parameters:

• model (type[BaseModelT]) –

  Config model type to parse TOML using.

• data (str) –

  TOML string to parse.

Returns:

• BaseModelT –

  Model initialized from TOML file.

Source code in academy/exchange/cloud/config.py

def loads(model: type[BaseModelT], data: str) -> BaseModelT:
    """Parse TOML string to data class.

    Args:
        model: Config model type to parse TOML using.
        data: TOML string to parse.

    Returns:
        Model initialized from TOML file.
    """
    data_dict = tomllib.loads(data)
    return model.model_validate(data_dict)