Skip to content

config

Configuration management module for Bridgic.

GlobalSetting

Bases: BaseModel

Global configuration settings for the Bridgic framework.

This class implements a singleton pattern to provide centralized configuration that applies across all Automa instances. The main methods are:

  • GlobalSetting.read(): Get the singleton global setting instance.
  • GlobalSetting.set(): Set the specific fields of the global setting instance.

Attributes:

Name Type Description
callback_builders List[WorkerCallbackBuilder]

Callback builders that will be automatically applied to all workers across all Automa instances.

Source code in bridgic/core/config/_global_setting.py
class GlobalSetting(BaseModel):
    """
    Global configuration settings for the Bridgic framework.

    This class implements a singleton pattern to provide centralized configuration
    that applies across all Automa instances. The main methods are:

    - `GlobalSetting.read()`: Get the singleton global setting instance.
    - `GlobalSetting.set()`: Set the specific fields of the global setting instance.

    Attributes
    ----------
    callback_builders : List[WorkerCallbackBuilder]
        Callback builders that will be automatically applied to all workers
        across all Automa instances.
    """
    model_config = {"arbitrary_types_allowed": True}

    callback_builders: List["WorkerCallbackBuilder"] = []
    """Global callback builders that will be applied to all workers."""

    # Singleton instance
    _instance: ClassVar[Optional["GlobalSetting"]] = None
    _lock: ClassVar[Lock] = Lock()

    @classmethod
    def read(cls) -> "GlobalSetting":
        """
        Get the singleton global setting instance.

        Returns
        -------
        GlobalSetting
            The singleton global setting instance.
        """
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = cls()
        return cls._instance

    @classmethod
    def set(
        cls,
        callback_builders: Optional[List["WorkerCallbackBuilder"]] = None,
    ) -> None:
        """
        Set global setting fields.

        This method allows you to update specific fields of the global setting
        without needing to create a complete GlobalSetting object.

        Parameters
        ----------
        callback_builders : Optional[List[WorkerCallbackBuilder]], optional
            Global callback builders that will be applied to all workers.
            If None, the current callback_builders are not changed.
        """
        instance = cls.read()
        with cls._lock:
            if callback_builders is not None:
                instance.callback_builders = callback_builders

    @classmethod
    def add(cls, callback_builder: Optional["WorkerCallbackBuilder"] = None) -> None:
        """
        Add new element to the existing field(s) of the `GlobalSetting`.

        Parameters
        ----------
        callback_builder : Optional[WorkerCallbackBuilder]
            The callback builder to add to the global setting callback builders. If None is passed in, nothing will be done.
        """
        instance = cls.read()
        with cls._lock:
            if callback_builder is not None:
                instance.callback_builders.append(callback_builder)

callback_builders class-attribute instance-attribute

callback_builders: List[WorkerCallbackBuilder] = []

Global callback builders that will be applied to all workers.

read

classmethod
read() -> GlobalSetting

Get the singleton global setting instance.

Returns:

Type Description
GlobalSetting

The singleton global setting instance.

Source code in bridgic/core/config/_global_setting.py
@classmethod
def read(cls) -> "GlobalSetting":
    """
    Get the singleton global setting instance.

    Returns
    -------
    GlobalSetting
        The singleton global setting instance.
    """
    if cls._instance is None:
        with cls._lock:
            if cls._instance is None:
                cls._instance = cls()
    return cls._instance

set

classmethod
set(
    callback_builders: Optional[
        List[WorkerCallbackBuilder]
    ] = None,
) -> None

Set global setting fields.

This method allows you to update specific fields of the global setting without needing to create a complete GlobalSetting object.

Parameters:

Name Type Description Default
callback_builders Optional[List[WorkerCallbackBuilder]]

Global callback builders that will be applied to all workers. If None, the current callback_builders are not changed.

None
Source code in bridgic/core/config/_global_setting.py
@classmethod
def set(
    cls,
    callback_builders: Optional[List["WorkerCallbackBuilder"]] = None,
) -> None:
    """
    Set global setting fields.

    This method allows you to update specific fields of the global setting
    without needing to create a complete GlobalSetting object.

    Parameters
    ----------
    callback_builders : Optional[List[WorkerCallbackBuilder]], optional
        Global callback builders that will be applied to all workers.
        If None, the current callback_builders are not changed.
    """
    instance = cls.read()
    with cls._lock:
        if callback_builders is not None:
            instance.callback_builders = callback_builders

add

classmethod
add(
    callback_builder: Optional[
        WorkerCallbackBuilder
    ] = None,
) -> None

Add new element to the existing field(s) of the GlobalSetting.

Parameters:

Name Type Description Default
callback_builder Optional[WorkerCallbackBuilder]

The callback builder to add to the global setting callback builders. If None is passed in, nothing will be done.

None
Source code in bridgic/core/config/_global_setting.py
@classmethod
def add(cls, callback_builder: Optional["WorkerCallbackBuilder"] = None) -> None:
    """
    Add new element to the existing field(s) of the `GlobalSetting`.

    Parameters
    ----------
    callback_builder : Optional[WorkerCallbackBuilder]
        The callback builder to add to the global setting callback builders. If None is passed in, nothing will be done.
    """
    instance = cls.read()
    with cls._lock:
        if callback_builder is not None:
            instance.callback_builders.append(callback_builder)

HttpClientConfig

Bases: TypedDict

Serializable configuration for creating an HTTP client.

This configuration can be serialized and deserialized, allowing the framework to recreate HTTP clients after serialization.

Attributes:

Name Type Description
headers Optional[Dict[str, str]]

HTTP headers to include with all requests.

timeout Optional[HttpClientTimeoutConfig]

Timeout configuration for the HTTP client.

auth Optional[HttpClientAuthConfig]

Authentication configuration for the HTTP client.

Source code in bridgic/core/config/_http_client_config.py
class HttpClientConfig(TypedDict, total=False):
    """
    Serializable configuration for creating an HTTP client.

    This configuration can be serialized and deserialized, allowing the framework
    to recreate HTTP clients after serialization.

    Attributes
    ----------
    headers : Optional[Dict[str, str]]
        HTTP headers to include with all requests.
    timeout : Optional[HttpClientTimeoutConfig]
        Timeout configuration for the HTTP client.
    auth : Optional[HttpClientAuthConfig]
        Authentication configuration for the HTTP client.
    """
    headers: Optional[Dict[str, str]]
    timeout: Optional[HttpClientTimeoutConfig]
    auth: Optional[HttpClientAuthConfig]

HttpClientTimeoutConfig

Bases: TypedDict

Configuration for HTTP client timeout settings.

Attributes:

Name Type Description
connect Optional[float]

Timeout for establishing a connection (seconds).

read Optional[float]

Timeout for reading data (seconds).

write Optional[float]

Timeout for writing data (seconds).

pool Optional[float]

Timeout for acquiring a connection from the pool (seconds).

Source code in bridgic/core/config/_http_client_config.py
class HttpClientTimeoutConfig(TypedDict, total=False):
    """
    Configuration for HTTP client timeout settings.

    Attributes
    ----------
    connect : Optional[float]
        Timeout for establishing a connection (seconds).
    read : Optional[float]
        Timeout for reading data (seconds).
    write : Optional[float]
        Timeout for writing data (seconds).
    pool : Optional[float]
        Timeout for acquiring a connection from the pool (seconds).
    """
    connect: Optional[float]
    read: Optional[float]
    write: Optional[float]
    pool: Optional[float]

HttpClientAuthConfig

Bases: TypedDict

Configuration for HTTP client authentication.

Attributes:

Name Type Description
type Literal['basic', 'bearer']

The type of authentication.

username Optional[str]

Username for basic auth (required if type is "basic").

password Optional[str]

Password for basic auth (required if type is "basic").

token Optional[str]

Bearer token (required if type is "bearer").

Source code in bridgic/core/config/_http_client_config.py
class HttpClientAuthConfig(TypedDict, total=False):
    """
    Configuration for HTTP client authentication.

    Attributes
    ----------
    type : Literal["basic", "bearer"]
        The type of authentication.
    username : Optional[str]
        Username for basic auth (required if type is "basic").
    password : Optional[str]
        Password for basic auth (required if type is "basic").
    token : Optional[str]
        Bearer token (required if type is "bearer").
    """
    type: Literal["basic", "bearer"]
    username: Optional[str]
    password: Optional[str]
    token: Optional[str]

create_http_client_from_config

create_http_client_from_config(
    config: Optional[HttpClientConfig],
    is_async: bool = True,
) -> Optional[Union[AsyncClient, Client]]

Create an HTTP client (sync or async) from a serializable configuration.

Parameters:

Name Type Description Default
config Optional[HttpClientConfig]

The HTTP client configuration. If None, returns None.

required
is_async bool

If True, creates an httpx.AsyncClient. If False, creates an httpx.Client. Defaults to True.

True

Returns:

Type Description
Optional[Union[AsyncClient, Client]]

The created HTTP client, or None if config is None.

Raises:

Type Description
ValueError

If the auth configuration is invalid (missing required fields or unsupported type).

Examples:

Create async client with custom headers:

1
2
3
4
>>> config = {
...     "headers": {"Authorization": "Bearer token123"}
... }
>>> client = create_http_client_from_config(config, is_async=True)

Create async client with custom timeout:

1
2
3
4
5
6
7
>>> config = {
...     "timeout": {
...         "connect": 10.0,
...         "read": 60.0
...     }
... }
>>> client = create_http_client_from_config(config, is_async=True)
Source code in bridgic/core/config/_http_client_config.py
def create_http_client_from_config(
    config: Optional[HttpClientConfig],
    is_async: bool = True,
) -> Optional[Union[httpx.AsyncClient, httpx.Client]]:
    """
    Create an HTTP client (sync or async) from a serializable configuration.

    Parameters
    ----------
    config : Optional[HttpClientConfig]
        The HTTP client configuration. If None, returns None.
    is_async : bool
        If True, creates an `httpx.AsyncClient`. If False, creates an `httpx.Client`.
        Defaults to True.

    Returns
    -------
    Optional[Union[httpx.AsyncClient, httpx.Client]]
        The created HTTP client, or None if config is None.

    Raises
    ------
    ValueError
        If the auth configuration is invalid (missing required fields or unsupported type).

    Examples
    --------
    Create async client with custom headers:
    >>> config = {
    ...     "headers": {"Authorization": "Bearer token123"}
    ... }
    >>> client = create_http_client_from_config(config, is_async=True)

    Create async client with custom timeout:
    >>> config = {
    ...     "timeout": {
    ...         "connect": 10.0,
    ...         "read": 60.0
    ...     }
    ... }
    >>> client = create_http_client_from_config(config, is_async=True)
    """
    # Return None when config is None.
    if config is None:
        return None

    # Extract configuration
    headers = config.get("headers")
    timeout_config = config.get("timeout")
    auth_config = config.get("auth")

    # Build timeout object if configured
    timeout = None
    if timeout_config:
        timeout = httpx.Timeout(
            connect=timeout_config.get("connect"),
            read=timeout_config.get("read"),
            write=timeout_config.get("write"),
            pool=timeout_config.get("pool"),
        )

    # Build auth object if configured
    auth = None
    if auth_config:
        auth_type = auth_config.get("type")
        if auth_type == "basic":
            username = auth_config.get("username")
            password = auth_config.get("password")
            if username is None or password is None:
                raise ValueError(
                    "Basic auth requires both 'username' and 'password' in http_client_config['auth']."
                )
            auth = httpx.BasicAuth(username=username, password=password)
        elif auth_type == "bearer":
            token = auth_config.get("token")
            if token is None:
                raise ValueError(
                    "Bearer auth requires 'token' in http_client_config['auth']."
                )
            # Bearer token is typically passed via headers.
            # So we add it to headers if not already present.
            if headers is None:
                headers = {}
            if "Authorization" not in headers:
                headers["Authorization"] = f"Bearer {token}"
        else:
            raise ValueError(
                f"Unsupported auth type: {auth_type}. Supported types: 'basic', 'bearer'."
            )

    # Build client kwargs
    client_kwargs: Dict[str, Any] = {
        "follow_redirects": True,
    }

    if timeout is not None:
        client_kwargs["timeout"] = timeout

    if headers is not None:
        client_kwargs["headers"] = headers

    if auth is not None:
        client_kwargs["auth"] = auth

    # Create the appropriate client type
    if is_async:
        return httpx.AsyncClient(**client_kwargs)
    else:
        return httpx.Client(**client_kwargs)