Last active
March 30, 2021 02:55
-
-
Save mpkocher/3836a9262cbbe2500c2496bdbf526429 to your computer and use it in GitHub Desktop.
MK common Python utils
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# Common core utils are too small that don't warrant creating a package | |
from argparse import ArgumentParser, Namespace, ArgumentDefaultsHelpFormatter | |
import csv | |
import functools | |
import logging | |
import sys | |
from typing import Callable as F | |
from typing import List, TypeVar, Type, Iterator | |
from pydantic import BaseModel | |
log = logging.getLogger(__name__) | |
B = TypeVar("B", bound=BaseModel) | |
def compose(*funcs): # type: ignore | |
"""Functional composition | |
[f, g, h] will be f(g(h(x))) | |
""" | |
def compose_two(f, g): # type: ignore | |
def c(x): # type: ignore | |
return f(g(x)) | |
return c | |
return functools.reduce(compose_two, funcs) | |
def setup_logger() -> None: | |
formatter = ( | |
"[%(levelname)s] %(asctime)-15sZ [%(name)s %(funcName)s %(lineno)d] %(message)s" | |
) | |
logging.basicConfig(level="DEBUG", stream=sys.stdout, format=formatter) | |
def add_opts_to_subcmd_1_parser(p: ArgumentParser) -> ArgumentParser: | |
return p | |
def run_sub_cmd_1(arg: Namespace) -> int: | |
return 0 | |
def get_parser_sp() -> ArgumentParser: | |
p = ArgumentParser(formatter_class=ArgumentDefaultsHelpFormatter) | |
p.add_argument("--version", action="version", version="0.1.0") | |
subparsers = p.add_subparsers() | |
def _add_to_sp( | |
name: str, | |
add_opts: F[[ArgumentParser], ArgumentParser], | |
func: F[[Namespace], int], | |
) -> ArgumentParser: | |
p = subparsers.add_parser(name, formatter_class=ArgumentDefaultsHelpFormatter) | |
add_opts(p) | |
p.set_defaults(func=func) | |
return p | |
_add_to_sp("subcmd-1", add_opts_to_subcmd_1_parser, run_sub_cmd_1) | |
return p | |
def run_main(argv: List[str]) -> int: | |
p = get_parser_sp() | |
pargs = p.parse_args(argv) | |
exit_code: int = pargs.func(pargs) | |
return exit_code | |
def csv_to_model(path: str, model_class: Type[B]) -> Iterator[B]: | |
with open(path) as csvfile: | |
reader = csv.DictReader(csvfile) | |
for row in reader: | |
yield model_class(**row) | |
class StrictModel(BaseModel): | |
class Config: | |
allow_mutation = False | |
validate_all = True | |
validate_assignment = True |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment