Config

dotflow.core.config.Config

Import

You can import the Config class with:

from dotflow import Config

from dotflow.providers import (
    StorageDefault,
    NotifyDefault,
    LogDefault,
    SchedulerDefault,
)

Example

class dotflow.core.config.Config

config = Config(
    storage=StorageFile(path=".output"),
    notify=NotifyDefault(),
    log=LogDefault()
)

Parameters:

Name	Type	Description	Default
storage	Optional[Storage]	Type of the storage.	None
notify	Optional[Notify]	Type of the notify.	None
log	Optional[Log]	Type of the log.	None
scheduler	Optional[Scheduler]	Type of the scheduler.	None
tracer	Optional[Tracer]	Type of the tracer.	None
metrics	Optional[Metrics]	Type of the metrics.	None

Attributes:

Name	Type	Description
storage	Optional[Storage]
notify	Optional[Notify]
log	Optional[Log]
scheduler	Optional[Scheduler]
tracer	Optional[Tracer]
metrics	Optional[Metrics]

Source code in dotflow/core/config.py

class Config:
    """
    Import:
        You can import the **Config** class with:

            from dotflow import Config

            from dotflow.providers import (
                StorageDefault,
                NotifyDefault,
                LogDefault,
                SchedulerDefault,
            )

    Example:
        `class` dotflow.core.config.Config

            config = Config(
                storage=StorageFile(path=".output"),
                notify=NotifyDefault(),
                log=LogDefault()
            )

    Args:
        storage (Optional[Storage]): Type of the storage.
        notify (Optional[Notify]): Type of the notify.
        log (Optional[Log]): Type of the log.
        scheduler (Optional[Scheduler]): Type of the scheduler.
        tracer (Optional[Tracer]): Type of the tracer.
        metrics (Optional[Metrics]): Type of the metrics.

    Attributes:
        storage (Optional[Storage]):
        notify (Optional[Notify]):
        log (Optional[Log]):
        scheduler (Optional[Scheduler]):
        tracer (Optional[Tracer]):
        metrics (Optional[Metrics]):
    """

    _PROVIDERS = {
        "storage": Storage,
        "notify": Notify,
        "log": Log,
        "server": Server,
        "scheduler": Scheduler,
        "tracer": Tracer,
        "metrics": Metrics,
    }

    def __init__(
        self,
        storage: Storage | None = None,
        notify: Notify | None = None,
        log: Log | None = None,
        server: Server | None = None,
        scheduler: Scheduler | None = None,
        tracer: Tracer | None = None,
        metrics: Metrics | None = None,
    ) -> None:
        self.storage = storage if storage is not None else StorageDefault()
        self.notify = notify if notify is not None else NotifyDefault()
        self.log = log if log is not None else LogDefault()
        self.server = server if server is not None else ServerDefault()
        self.scheduler = (
            scheduler if scheduler is not None else SchedulerDefault()
        )
        self.tracer = tracer if tracer is not None else TracerDefault()
        self.metrics = metrics if metrics is not None else MetricsDefault()

        self._validate()

    def _validate(self) -> None:
        for name, abc in self._PROVIDERS.items():
            value = getattr(self, name)
            if value is not None and not isinstance(value, abc):
                raise NotCallableObject(name=name)

storage = storage if storage is not None else StorageDefault() instance-attribute

notify = notify if notify is not None else NotifyDefault() instance-attribute

log = log if log is not None else LogDefault() instance-attribute

server = server if server is not None else ServerDefault() instance-attribute

scheduler = scheduler if scheduler is not None else SchedulerDefault() instance-attribute

tracer = tracer if tracer is not None else TracerDefault() instance-attribute

metrics = metrics if metrics is not None else MetricsDefault() instance-attribute