modules
Top-level package for Cfg4Py.
cli
¶
Command
¶
Source code in cfg4py/cli.py
class Command:
def __init__(self):
self.resource_path = os.path.normpath(
os.path.join(os.path.dirname(__file__), "resources/")
)
self.yaml = YAML(typ="safe") # default, if not specfied, is 'rt' (round-trip)
self.yaml.default_flow_style = False
with open(
os.path.join(self.resource_path, "template.yaml"), "r", encoding="utf-8"
) as f:
self.templates = self.yaml.load(f)
self.transformed = self._transform()
def build(self, config_dir: str):
"""Compile configuration files into python script, which is used by IDE's
auto-complete function
Args:
config_dir: The folder where your configuration files located
Returns:
"""
if not os.path.exists(config_dir):
print(f"path {config_dir} not exists")
sys.exit(-1)
count = 0
for f in os.listdir(config_dir):
if (
f.startswith("default")
or f.startswith("dev")
or f.startswith("test")
or f.startswith("production")
):
print(f"found {f}")
count += 1
if count > 0:
print(f"{count} files found in total")
else:
print("the folder contains no valid configuration files")
sys.exit(-1)
try:
init(config_dir)
sys.path.insert(0, config_dir)
from schema import Config # type: ignore # noqa
output_file = f"{os.path.join(config_dir, 'schema')}"
msg = f"Config file is built with success and saved at {output_file}"
print(msg)
except Exception as e: # pragma: no cover
logging.exception(e)
print("Config file built failure.")
def _choose_dest_dir(self, dst):
if dst is None:
dst = input("Where should I save configuration files?\n")
if os.path.exists(dst):
for f in os.listdir(dst):
msg = f"The folder already contains {f}, please choose clean one."
if f in ["defaults.yaml", "dev.yaml", "test.yaml", "production.yaml"]:
print(msg)
return None
return dst
else:
create = input("Path not exists, create('Q' to exit)? [Y/n]")
if create.upper() == "Y":
os.makedirs(dst, exist_ok=True)
return dst
elif create.upper() == "Q":
sys.exit(-1)
else:
return None
def scaffold(self, dst: Optional[str]):
"""Creates initial configuration files based on our choices.
Args:
dst:
Returns:
"""
print("Creating a configuration boilerplate:")
dst = self._choose_dest_dir(dst)
while dst is None:
dst = self._choose_dest_dir(dst)
yaml = YAML(typ="safe") # default, if not specfied, is 'rt' (round-trip)
with open(
os.path.join(self.resource_path, "template.yaml"), "r", encoding="utf-8"
) as f:
templates = yaml.load(f)
print("Which flavors do you want?")
print("-" * 20)
prompt = """
0 - console + rotating file logging
10 - redis/redis-py (gh://andymccurdy/redis-py)
11 - redis/aioredis (gh://aio-libs/aioredis)
20 - mysql/PyMySQL (gh://PyMySQL/PyMySQL)
30 - postgres/asyncpg (gh://MagicStack/asyncpg)
31 - postgres/psycopg2 (gh://psycopg/psycopg2)
40 - mq/pika (gh://pika/pika)
50 - mongodb/pymongo (gh://mongodb/mongo-python-driver)
"""
print(prompt)
chooses = input(
"Please choose flavors by index, separated each by a comma(,):\n"
)
flavors = {}
mapping = {
"0": "logging",
"1": "redis",
"2": "mysql",
"3": "postgres",
"4": "mq",
"5": "mongodb",
}
for index in chooses.strip(" ").split(","):
if index == "0":
flavors["logging"] = templates["logging"]
continue
try:
major = mapping[index[0]]
minor = int(index[1])
flavors[major] = list(templates[major][minor].values())[0]
except (ValueError, KeyError):
print(f"Wrong index {index}, skipped.")
continue
with open(os.path.join(dst, "defaults.yaml"), "w", encoding="utf-8") as f:
f.writelines(
"#auto generated by Cfg4Py: https://github.com/jieyu-tech/cfg4py\n"
)
yaml.dump(flavors, f)
print(f"Cfg4Py has generated the following files under {dst}:")
print("defaults.yaml")
for name in ["dev.yaml", "test.yaml", "production.yaml"]:
with open(os.path.join(dst, name), "w", encoding="utf8") as f:
f.writelines(
"#auto generated by Cfg4Py: https://github.com/jieyu-tech/cfg4py\n"
)
print(name)
with open(os.path.join(dst, "defaults.yaml"), "r", encoding="utf-8") as f:
print("Content in defaults.yaml")
for line in f.readlines():
print(line.replace("\n", ""))
def _show_supported_config(self):
print("Support the following configurations:")
for key in self.templates.keys():
item = self.templates.get(key)
if isinstance(item, dict):
print(f" {key}")
elif isinstance(item, list):
sub_keys = []
for sub_item in item:
sub_keys.append(f"{key}/{list(sub_item.keys())[0]}")
print(f" {key}: {', '.join(sub_keys)}")
def _transform(self):
transformed = {}
for key in self.templates:
if isinstance(self.templates[key], dict):
transformed[key] = self.templates[key]
elif isinstance(self.templates[key], list):
for item in self.templates[key]:
item_key = list(item.keys())[0]
transformed[f"{key}/{item_key}"] = item
return transformed
def hint(self, what: str = None, usage: bool = False):
"""show a cheat sheet for configurations.
for example:
cfg4py hint mysql
this will print how to configure PyMySQL
:param what
:param usage
"""
if what is None or (
(what not in self.templates) and what not in self.transformed
):
return self._show_supported_config()
usage_key = f"{what}_usage"
if usage_key in self.templates and usage:
print("Usage:", self.templates.get(usage_key))
if what in self.templates:
self.yaml.dump(self.templates[what], sys.stdout)
if usage_key in self.transformed and usage:
print("Usage:", self.transformed.get(usage_key))
if what in self.transformed:
self.yaml.dump(self.transformed[what], sys.stdout)
def set_server_role(self):
print("please add the following line into your .bashrc:\n")
print(f"export {envar}=DEV\n")
msg = "You need to change DEV to TEST | PRODUCTION according to its actual role\
accordingly"
print(msg)
def version(self):
from cfg4py import __version__
print(
"Easy config module support code complete, cascading design and apative deployment"
)
print(f"version: {__version__}")
build(self, config_dir)
¶
Compile configuration files into python script, which is used by IDE's auto-complete function
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config_dir |
str |
The folder where your configuration files located |
required |
Source code in cfg4py/cli.py
def build(self, config_dir: str):
"""Compile configuration files into python script, which is used by IDE's
auto-complete function
Args:
config_dir: The folder where your configuration files located
Returns:
"""
if not os.path.exists(config_dir):
print(f"path {config_dir} not exists")
sys.exit(-1)
count = 0
for f in os.listdir(config_dir):
if (
f.startswith("default")
or f.startswith("dev")
or f.startswith("test")
or f.startswith("production")
):
print(f"found {f}")
count += 1
if count > 0:
print(f"{count} files found in total")
else:
print("the folder contains no valid configuration files")
sys.exit(-1)
try:
init(config_dir)
sys.path.insert(0, config_dir)
from schema import Config # type: ignore # noqa
output_file = f"{os.path.join(config_dir, 'schema')}"
msg = f"Config file is built with success and saved at {output_file}"
print(msg)
except Exception as e: # pragma: no cover
logging.exception(e)
print("Config file built failure.")
hint(self, what=None, usage=False)
¶
show a cheat sheet for configurations. for example: cfg4py hint mysql this will print how to configure PyMySQL :param what :param usage
Source code in cfg4py/cli.py
def hint(self, what: str = None, usage: bool = False):
"""show a cheat sheet for configurations.
for example:
cfg4py hint mysql
this will print how to configure PyMySQL
:param what
:param usage
"""
if what is None or (
(what not in self.templates) and what not in self.transformed
):
return self._show_supported_config()
usage_key = f"{what}_usage"
if usage_key in self.templates and usage:
print("Usage:", self.templates.get(usage_key))
if what in self.templates:
self.yaml.dump(self.templates[what], sys.stdout)
if usage_key in self.transformed and usage:
print("Usage:", self.transformed.get(usage_key))
if what in self.transformed:
self.yaml.dump(self.transformed[what], sys.stdout)
scaffold(self, dst)
¶
Creates initial configuration files based on our choices.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
dst |
Optional[str] |
required |
Source code in cfg4py/cli.py
def scaffold(self, dst: Optional[str]):
"""Creates initial configuration files based on our choices.
Args:
dst:
Returns:
"""
print("Creating a configuration boilerplate:")
dst = self._choose_dest_dir(dst)
while dst is None:
dst = self._choose_dest_dir(dst)
yaml = YAML(typ="safe") # default, if not specfied, is 'rt' (round-trip)
with open(
os.path.join(self.resource_path, "template.yaml"), "r", encoding="utf-8"
) as f:
templates = yaml.load(f)
print("Which flavors do you want?")
print("-" * 20)
prompt = """
0 - console + rotating file logging
10 - redis/redis-py (gh://andymccurdy/redis-py)
11 - redis/aioredis (gh://aio-libs/aioredis)
20 - mysql/PyMySQL (gh://PyMySQL/PyMySQL)
30 - postgres/asyncpg (gh://MagicStack/asyncpg)
31 - postgres/psycopg2 (gh://psycopg/psycopg2)
40 - mq/pika (gh://pika/pika)
50 - mongodb/pymongo (gh://mongodb/mongo-python-driver)
"""
print(prompt)
chooses = input(
"Please choose flavors by index, separated each by a comma(,):\n"
)
flavors = {}
mapping = {
"0": "logging",
"1": "redis",
"2": "mysql",
"3": "postgres",
"4": "mq",
"5": "mongodb",
}
for index in chooses.strip(" ").split(","):
if index == "0":
flavors["logging"] = templates["logging"]
continue
try:
major = mapping[index[0]]
minor = int(index[1])
flavors[major] = list(templates[major][minor].values())[0]
except (ValueError, KeyError):
print(f"Wrong index {index}, skipped.")
continue
with open(os.path.join(dst, "defaults.yaml"), "w", encoding="utf-8") as f:
f.writelines(
"#auto generated by Cfg4Py: https://github.com/jieyu-tech/cfg4py\n"
)
yaml.dump(flavors, f)
print(f"Cfg4Py has generated the following files under {dst}:")
print("defaults.yaml")
for name in ["dev.yaml", "test.yaml", "production.yaml"]:
with open(os.path.join(dst, name), "w", encoding="utf8") as f:
f.writelines(
"#auto generated by Cfg4Py: https://github.com/jieyu-tech/cfg4py\n"
)
print(name)
with open(os.path.join(dst, "defaults.yaml"), "r", encoding="utf-8") as f:
print("Content in defaults.yaml")
for line in f.readlines():
print(line.replace("\n", ""))
core
¶
Main module.
LocalConfigChangeHandler (FileSystemEventHandler)
¶
Source code in cfg4py/core.py
class LocalConfigChangeHandler(FileSystemEventHandler):
def dispatch(self, event):
if not isinstance(event, FileModifiedEvent):
return
ext = os.path.splitext(event.src_path)
if ext in [".yml", ".yaml"]:
_load_from_local_file()
dispatch(self, event)
¶
Dispatches events to the appropriate methods.
:param event:
The event object representing the file system event.
:type event:
:class:FileSystemEvent
Source code in cfg4py/core.py
def dispatch(self, event):
if not isinstance(event, FileModifiedEvent):
return
ext = os.path.splitext(event.src_path)
if ext in [".yml", ".yaml"]:
_load_from_local_file()
config_remote_fetcher(fetcher, interval=300)
¶
config a remote configuration fetcher, which will pull the settings on every
refresh_interval
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fetcher |
RemoteConfigFetcher |
sub class of |
required |
interval |
int |
how long should cfg4py to pull the configuration from remote |
300 |
Source code in cfg4py/core.py
def config_remote_fetcher(fetcher: RemoteConfigFetcher, interval: int = 300):
"""
config a remote configuration fetcher, which will pull the settings on every
`refresh_interval`
Args:
fetcher: sub class of `RemoteConfigFetcher`
interval: how long should cfg4py to pull the configuration from remote
Returns:
"""
global _remote_fetcher
_remote_fetcher = fetcher
_scheduler.add_job(_refresh, "interval", seconds=interval)
_scheduler.start()
enable_logging(level=20, log_file=None, file_size=10, file_count=7)
¶
Enable basic log function for the application
if log_file is None, then it'll provide console logging, otherwise, the console logging is turned off, all events will be logged into the provided file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
level |
the log level, one of logging.DEBUG, logging.INFO, logging.WARNING, |
20 |
|
log_file |
the absolute file path for the log. |
None |
|
file_size |
file size in MB unit |
10 |
|
file_count |
how many backup files leaved in disk |
7 |
Returns:
| Type | Description |
|---|---|
None |
Source code in cfg4py/core.py
def enable_logging(level=logging.INFO, log_file=None, file_size=10, file_count=7):
"""
Enable basic log function for the application
if log_file is None, then it'll provide console logging, otherwise, the console
logging is turned off, all events will be logged into the provided file.
Args:
level: the log level, one of logging.DEBUG, logging.INFO, logging.WARNING,
logging.Error
log_file: the absolute file path for the log.
file_size: file size in MB unit
file_count: how many backup files leaved in disk
Returns:
None
"""
assert file_count > 0
assert file_size > 0
from logging import handlers
formatter = logging.Formatter(
"%(asctime)s %(levelname)-1.1s %(filename)s:%(lineno)s | %(message)s"
)
_logger = logging.getLogger()
_logger.setLevel(level)
if log_file is None:
console = logging.StreamHandler()
console.setFormatter(formatter)
_logger.addHandler(console)
else:
file_dir = os.path.dirname(log_file)
os.makedirs(file_dir, exist_ok=True)
rotating_file = handlers.RotatingFileHandler(
log_file, maxBytes=1024 * 1024 * file_size, backupCount=file_count
)
rotating_file.setFormatter(formatter)
_logger.addHandler(rotating_file)
init(local_cfg_path=None, dump_on_change=True, strict=False)
¶
create cfg object.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
local_cfg_path |
str |
the directory name where your configuration files exist |
None |
dump_on_change |
if configuration is updated, whether or not to dump them into log file |
True |
Source code in cfg4py/core.py
def init(local_cfg_path: str = None, dump_on_change=True, strict=False):
"""
create cfg object.
Args:
local_cfg_path: the directory name where your configuration files exist
dump_on_change: if configuration is updated, whether or not to dump them into
log file
Returns:
"""
global _local_config_dir, _dump_on_change, _remote_fetcher, _local_observer
global _cfg_obj, _cfg_local, _cfg_remote
global _strict
_strict = strict
_dump_on_change = dump_on_change
if local_cfg_path:
_local_config_dir = os.path.expanduser(local_cfg_path)
_cfg_local = _load_from_local_file()
update_config(_mixin(_cfg_remote, _cfg_local))
try:
# handle local configuration file change, this may not be available on some platform, like apple m1
_local_observer = Observer()
_local_observer.schedule(
LocalConfigChangeHandler(), _local_config_dir, recursive=False
)
_local_observer.start()
except Exception as e:
logger.exception(e)
logger.warning("failed to watch file changes. Hot-reload is not available")
return _cfg_obj