Marmelator/pyinfra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 36 Wiki Activity

RED-4653: Reworked the configuration as a plain python object

Browse Source
This commit is contained in:
Viktor Seifert 2022-07-15 14:21:13 +02:00
parent 5f97b3dedb
commit 15b23a0019
1 changed files with 38 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
pyinfra/config.py
View File

@ -1,55 +1,54 @@
"""Implements a config object with dot-indexing syntax.""" from os import environ
import os
from itertools import chain
from operator import truth
from envyaml import EnvYAML
from funcy import first, juxt, butlast, last
from pyinfra.locations import CONFIG_FILE
def _get_item_and_maybe_make_dotindexable(container, item): def read_from_environment(environment_variable_name, default_value):
ret = container[item] return environ.get(environment_variable_name, default_value)
return DotIndexable(ret) if isinstance(ret, dict) else ret
class DotIndexable: class Config(object):
def __init__(self, x): def __init__(self):
self.x = x # Logging level for service logger
self.logging_level_root = read_from_environment("LOGGING_LEVEL_ROOT", "DEBUG")
def __getattr__(self, item): # RabbitMQ host address
return _get_item_and_maybe_make_dotindexable(self.x, item) self.rabbitmq_host = read_from_environment("RABBITMQ_HOST", "localhost")
def __repr__(self): # RabbitMQ host port
return self.x.__repr__() self.rabbitmq_port = read_from_environment("RABBITMQ_PORT", "5672")
def __getitem__(self, item): # RabbitMQ username
return self.__getattr__(item) self.rabbitmq_username = read_from_environment("RABBITMQ_USERNAME", "user")
# RabbitMQ password
self.rabbitmq_password = read_from_environment("RABBITMQ_PASSWORD", "bitnami")
class Config: # Controls AMQP heartbeat timeout in seconds
def __init__(self, config_path): self.rabbitmq_heartbeat = read_from_environment("RABBITMQ_HEARTBEAT", "7200")
self.__config = EnvYAML(config_path)
def __getattr__(self, item): # Queue name for requests to the service
if item in self.__config: self.request_queue = read_from_environment("REQUEST_QUEUE", "request_queue")
return _get_item_and_maybe_make_dotindexable(self.__config, item)
def __getitem__(self, item): # Queue name for responses by service
return self.__getattr__(item) self.response_queue = read_from_environment("RESPONSE_QUEUE", "response_queue")
# Queue name for failed messages
self.dead_letter_queue = read_from_environment("DEAD_LETTER_QUEUE", "dead_letter_queue")
CONFIG = Config(CONFIG_FILE) # The type of storage to use {s3, azure}
self.storage_backend = read_from_environment("STORAGE_BACKEND", "s3")
# The bucket / container to pull files specified in queue requests from
self.storage_bucket = read_from_environment("STORAGE_BUCKET", "pyinfra-test-bucket")
def parse_disjunction_string(disjunction_string): # Endpoint for s3 storage
def try_parse_env_var(disjunction_string): self.storage_endpoint = read_from_environment("STORAGE_ENDPOINT", "http://127.0.0.1:9000")
try:
return os.environ[disjunction_string]
except KeyError:
return None
options = disjunction_string.split("|") # User for s3 storage
identifiers, fallback_value = juxt(butlast, last)(options) self.storage_key = read_from_environment("STORAGE_KEY", "root")
return first(chain(filter(truth, map(try_parse_env_var, identifiers)), [fallback_value]))
# Password for s3 storage
self.storage_secret = read_from_environment("STORAGE_SECRET", "password")
# Connection string for Azure storage
self.storage_azureconnectionstring = read_from_environment("STORAGE_AZURECONNECTIONSTRING",
"DefaultEndpointsProtocol=...")