pydantic based configuration wrangler. Handles reading multiple ini or toml files with inheritance rules and variable expansions.
Project description
Config Wrangler
pydantic based configuration wrangler. Handles reading multiple ini or toml files with inheritance rules and variable expansions.
Installation
Install using your package manager of choice:
poetry add config-wrangler
pip install -U config-wrangler
conda install config-wrangler -c conda-forge
.
A Simple Example
config.ini
[S3_Source]
bucket_name=my.exmple-bucket
key_prefixes=processed/
user_id=AK123456789ABC
# Not a secure way to store the password, but OK for local prototype or examples.
# See KEYRING or KEEPASS for better options
password_source=CONFIG_FILE
raw_password=My secret password
[target_database]
dialect=sqlite
database_name=${test_section:my_environment:source_data_dir}/example_db
[test_section]
my_int=123
my_float=123.45
my_bool=Yes
my_str=ABC☕
my_bytes=ABCⓁⓄⓋ☕
my_list_auto_c=a,b,c
my_list_auto_nl=
a
b
c
my_list_auto_pipe=a|b|c
my_list_c=a,b,c
my_list_python=['x','y','z']
my_list_json=["J","S","O","N"]
my_list_nl=
a
b
c
my_list_int_c=1,2,3
my_tuple_c=a,b,c
my_tuple_nl=
a
b
c
my_tuple_int_c=1,2,3
my_dict={1: "One", 2: "Two"}
my_dict_str_int={"one": 1, "two": 2}
my_set={'A','B','C'}
my_set_int=1,2,3
my_frozenset=A,B,C
my_date=2021-05-31
my_time=11:55:23
my_datetime=2021-05-31 11:23:53
my_url=https://localhost:6553/
[test_section.my_environment]
name=dev
# For example to run we'll make both paths relative to current
temp_data_dir=.\temp_data\${test_section:my_environment:name}
source_data_dir=.
python code
import typing
from datetime import date, time, datetime
from pydantic import BaseModel, DirectoryPath, Field, AnyHttpUrl
from config_wrangler.config_data_loaders.base_config_data_loader import BaseConfigDataLoader
from config_wrangler.config_from_ini_env import ConfigFromIniEnv
from config_wrangler.config_from_loaders import ConfigFromLoaders
from config_wrangler.config_templates.config_hierarchy import ConfigHierarchy
from config_wrangler.config_templates.aws.s3_bucket import S3_Bucket
from config_wrangler.config_templates.sqlalchemy_database import SQLAlchemyDatabase
from config_wrangler.config_types.path_types import AutoCreateDirectoryPath
from config_wrangler.config_types.delimited_field import DelimitedListField
class S3_Bucket_KeyPrefixes(S3_Bucket):
key_prefixes: typing.List[str]
class Environment(ConfigHierarchy):
name: str = Field(..., env='env_name')
temp_data_dir: AutoCreateDirectoryPath
source_data_dir: DirectoryPath
class TestSection(BaseModel):
my_int: int
my_float: float
my_bool: bool
my_str: str
my_bytes: bytes
my_list_auto_c: list
my_list_auto_nl: list
my_list_auto_pipe: list
my_list_python: list
my_list_json: list
my_list_c: list = DelimitedListField(delimiter=',')
my_list_nl: list = DelimitedListField(delimiter='\n')
my_list_int_c: typing.List[int] = DelimitedListField(delimiter=',')
my_tuple_c: tuple = DelimitedListField(delimiter=',')
my_tuple_nl: tuple = DelimitedListField(delimiter='\n')
my_tuple_int_c: typing.Tuple[int, int, int] = DelimitedListField(delimiter=',')
my_dict: dict
my_dict_str_int: typing.Dict[str, int]
my_set: set
my_set_int: typing.Set[int]
my_frozenset: frozenset
my_date: date
my_time: time
my_datetime: datetime
my_url: AnyHttpUrl
my_environment: Environment
class ETLConfig(ConfigFromIniEnv):
class Config:
validate_default = True
validate_assignment = True
target_database: SQLAlchemyDatabase
s3_source: S3_Bucket_KeyPrefixes
test_section: TestSection
class ETLConfigAnyLoaders(ETLConfig):
def __init__(
self,
_config_data_loaders: typing.List[BaseConfigDataLoader],
**kwargs: typing.Dict[str, typing.Any]
) -> None:
# Skip super and call the next higher class
ConfigFromLoaders.__init__(
self,
_config_data_loaders=_config_data_loaders,
**kwargs
)
def main():
config = ETLConfig(file_name='simple_example.ini')
print(f"Temp data dir = {config.test_section.my_environment.temp_data_dir}")
# > Temp data dir = temp_data\dev
print(f"Source data dir = {config.test_section.my_environment.source_data_dir}")
# > Source data dir = .
print(f"my_int = {config.test_section.my_int}")
# > my_int = 123
print(f"my_float = {config.test_section.my_float}")
# > my_float = 123.45
print(f"my_str = {config.test_section.my_str}")
# > my_str = ABC☕
print(f"my_list_auto_c = {config.test_section.my_list_auto_c}")
# > my_list_auto_c = ['a', 'b', 'c']
print(f"my_list_auto_nl = {config.test_section.my_list_auto_nl}")
# > my_list_auto_c = ['a', 'b', 'c']
print(f"my_dict = {config.test_section.my_dict}")
# > my_dict = {1: 'One', 2: 'Two'}
print(f"my_set = {config.test_section.my_set}")
# > my_set = {'C', 'A', 'B'}
print(f"my_time = {config.test_section.my_time}")
# > my_time = 11:55:23
print(f"my_datetime = {config.test_section.my_datetime}")
# > my_datetime = 2021-05-31 11:23:53
print(f"my_url = {config.test_section.my_url}")
# > my_url = https://localhost:6553/
# Getting DB engine (requires sqlalchemy optional install
engine = config.target_database.get_engine()
print(f"target_database.engine = {engine}")
# > target_database.engine = Engine(sqlite:///.example_db)
print("Getting S3 Data")
bucket = config.s3_source.get_bucket()
print(f"S3 bucket definition = {bucket}")
for prefix in config.s3_source.key_prefixes:
print(f" bucket search prefix = {prefix}")
# > Getting S3 Data
# > credentials.py:56: UserWarning: Passwords stored directly in config or worse in code are not safe. Please make sure to fix this before deploying.
# > S3 bucket definitition = s3.Bucket(name='my.exmple-bucket')
# > bucket search prefix = processed/
if __name__ == '__main__':
main()
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
config_wrangler-1.2.0a0.tar.gz
(199.1 kB
view hashes)
Built Distribution
Close
Hashes for config_wrangler-1.2.0a0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 4df35b5298ddbb8598b7600f596a38459e436302f7899a92386ffff1d0f61e44 |
|
MD5 | 511f35f2199a375f477515f68f1eaca8 |
|
BLAKE2b-256 | bea427324707d9dfe53e5b53ce85fcdd3e36b214918f3b267df63309faf0e89e |