dlt.common.configuration.specs.pluggable_run_context
RunContextBase Objects
class RunContextBase(ABC)
Describes where dlt looks for settings, pipeline working folder. Implementations must be picklable.
__init__
@abstractmethod
def __init__(run_dir: Optional[str], *args: Any, **kwargs: Any)
An explicit run_dir, if None, run_dir should be auto-detected by particular implementation
name
@property
@abstractmethod
def name() -> str
Name of the run context. Entities like sources and destinations added to registries when this context
is active, will be scoped to it. Typically corresponds to Python package name ie. dlt.
uri
@property
@abstractmethod
def uri() -> str
Uniquely identifies the context. By default it is a combination of run_dir and runtime_kwargs
to create file:// uri
global_dir
@property
@abstractmethod
def global_dir() -> str
Directory in which global settings are stored ie ~/.dlt/
run_dir
@property
@abstractmethod
def run_dir() -> str
Defines the context working directory, defaults to cwd()
local_dir
@property
@abstractmethod
def local_dir() -> str
Defines data dir where local relative dirs and files are created, defaults to run_dir
settings_dir
@property
@abstractmethod
def settings_dir() -> str
Defines where the current settings (secrets and configs) are located
data_dir
@property
@abstractmethod
def data_dir() -> str
Defines where the pipelines working folders are stored.
module
@property
@abstractmethod
def module() -> Optional[ModuleType]
if run_dir is a top level importable python module, returns it, otherwise return None
runtime_kwargs
@property
@abstractmethod
def runtime_kwargs() -> Dict[str, Any]
Additional kwargs used to initialize this instance of run context, used for reloading
initial_providers
@abstractmethod
def initial_providers() -> List[ConfigProvider]
Returns initial providers for this context
initialize_runtime
@abstractmethod
def initialize_runtime(runtime_config: RuntimeConfiguration = None) -> None
Initializes runtime (ie. log, telemetry) using RuntimeConfiguration
runtime_config
@property
@abstractmethod
def runtime_config() -> RuntimeConfiguration
Runtime configuration used for initialize_runtime
config
@property
@abstractmethod
def config() -> BaseConfiguration
Returns (optionally resolves) run context configuration
get_data_entity
@abstractmethod
def get_data_entity(entity: str) -> str
Gets path in data_dir where entity (ie. pipelines, repos) are stored
get_run_entity
@abstractmethod
def get_run_entity(entity: str) -> str
Gets path in run_dir where entity (ie. sources, destinations etc.) are stored
get_setting
@abstractmethod
def get_setting(setting_path: str) -> str
Gets path in settings_dir where setting (ie. secrets.toml) are stored
unplug
@abstractmethod
def unplug() -> None
Called when context removed from container
plug
@abstractmethod
def plug() -> None
Called when context is added to container
reload
def reload() -> "RunContextBase"
This will reload current context by triggering run context plugin via Container
import_run_dir_module
@staticmethod
def import_run_dir_module(run_dir: str) -> ModuleType
Returns a top Python module of the workspace (if importable)
reset_config
@abstractmethod
def reset_config() -> None
Hook for contexts that store resolved configuration to reset it
ProfilesRunContext Objects
class ProfilesRunContext(RunContextBase)
Adds profile support on run context. Note: runtime checkable protocols are slow on isinstance
profile
@property
@abstractmethod
def profile() -> str
Returns current profile name
default_profile
@property
@abstractmethod
def default_profile() -> str
Returns default profile name
available_profiles
@abstractmethod
def available_profiles() -> List[str]
Returns available profiles
configured_profiles
def configured_profiles() -> List[str]
Returns profiles with configurations or dlt entities, same as available by default
switch_profile
@abstractmethod
def switch_profile(new_profile: str) -> Self
Switches current profile and returns new run context
PluggableRunContext Objects
class PluggableRunContext(ContainerInjectableContext)
Injectable run context taken via plugin
reload
def reload(run_dir_or_context: Optional[Union[str, RunContextBase]] = None,
runtime_kwargs: Dict[str, Any] = None) -> None
Reloads the context, using existing settings if not overwritten with method args
initialize_runtime
def initialize_runtime() -> None
Calls initialize_runtime on context only if active in container. We do not want to initialize runtime if instance is not active
push_context
def push_context() -> str
Pushes current context on stack and returns assert cookie
pop_context
def pop_context(cookie: str) -> None
Pops context from stack and re-initializes it if in container
drop_context
def drop_context(cookie: str) -> None
Pops context form stack but leaves new context for good