first commit

Maclovi · May 1, 2024 · 4de6e7b · 4de6e7b
commit 4de6e7b
Show file tree

Hide file tree

Showing 6 changed files with 311 additions and 0 deletions.
diff --git a/.gitignore b/.gitignore
@@ -0,0 +1,36 @@
+# Byte-compiled / optimized / DLL files
+**__pycache__/
+*.py[cod]
+*$py.class
+
+# Distribution / packaging
+MANIFEST
+
+# Unit test / coverage reports
+.pytest_cache/
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# ruff
+.ruff_cache
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
diff --git a/README.md b/README.md
@@ -0,0 +1,4 @@
+## Нука
+
+- зачем
+- что и почему
diff --git a/pyproject.toml b/pyproject.toml
@@ -0,0 +1,76 @@
+[project]
+name = "envmapp"
+dynamic = ["version"]
+description = "It's a simple mapping your environment"
+readme = "README.md"
+requires-python = ">=3.8"
+authors = [
+    { name = "Sergey Yavorsky", email = "yavorskyserge@gmail.com" },
+]
+classifiers = [
+]
+dependencies = [
+]
+
+[project.urls]
+Homepage = ""
+Documentation = ""
+Repository = ""
+
+[project.optional-dependencies]
+
+standard = [
+]
+
+all = [
+]
+
+[tool.mypy]
+strict = true
+
+[tool.pytest.ini_options]
+addopts = [
+  "--strict-config",
+  "--strict-markers",
+  "--ignore=docs_src",
+]
+xfail_strict = true
+junit_family = "xunit2"
+filterwarnings = [
+]
+
+[tool.ruff]
+line-length = 80
+
+[tool.ruff.lint]
+select = [
+    "E",  # pycodestyle errors
+    "W",  # pycodestyle warnings
+    "F",  # pyflakes
+    "I",  # isort
+    "B",  # flake8-bugbear
+    "C4",  # flake8-comprehensions
+    "UP",  # pyupgrade
+]
+ignore = [
+    "B008",  # do not perform function calls in argument defaults
+    "C901",  # too complex
+    "W191",  # indentation contains tabs
+]
+
+[tool.ruff.lint.pydocstyle]
+convention = "google"
+
+[tool.ruff.lint.per-file-ignores]
+"__init__.py" = ["F401"]
+
+[tool.ruff.lint.isort]
+known-third-party = ["fastapi", "pydantic", "starlette"]
+
+[tool.ruff.lint.pyupgrade]
+# Preserve types, even if a file imports `from __future__ import annotations`.
+keep-runtime-typing = true
+
+[tool.isort]
+profile = "black"
+line_length = 80
diff --git a/src/envmapp/dotenv.py b/src/envmapp/dotenv.py
@@ -0,0 +1,133 @@
+import logging
+import os
+from functools import wraps
+from pathlib import Path
+from typing import Any, Callable, ParamSpec, get_args, get_origin
+
+log = logging.getLogger(__name__)
+
+F_Spec = ParamSpec("F_Spec")
+F_Return = Any
+EnvMap = dict[str, str]
+
+
+class MetaClass(type):
+    def __new__(
+        cls,
+        name: str,
+        bases: tuple[Any],
+        namespace: dict[str, Any],
+        **kwargs: Any,
+    ) -> type:
+        if kwargs:
+            pass
+        return super().__new__(cls, name, bases, namespace)
+
+    def __init__(
+        cls,
+        name: str,
+        bases: tuple[Any],
+        namespace: dict[str, Any],
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(name, bases, namespace)
+
+        cls._environ_from_file: dict[str, Any] = {}
+        cls.path_env: str = kwargs.get("load_env", "")
+        cls.override: bool = kwargs.get("override", False)
+
+    def __call__(cls, *args: Any, **kwargs: Any) -> Any:
+        instance = super().__call__(*args, **kwargs)
+        cls.load_to_env(instance)
+        cls.create_const(instance)
+        return instance
+
+    def load_to_env(cls, instance: "MetaClass") -> None:
+        if os.path.isfile(cls.path_env):
+            cls._load_to_env(instance)
+        elif cls.path_env:
+            raise FileNotFoundError(
+                f"No such file - {cls.path_env!r}\n"
+                f"Your current dir is - {Path(__file__).parent.resolve()}"
+            )
+        elif cls.override:
+            log.warning("\tNothing to override!")
+
+    def _load_to_env(cls, instance: "MetaClass") -> None:
+        if not cls._environ_from_file:
+            env: "EnvMap" = get_from_file_env(cls.path_env)
+            cls._environ_from_file |= {
+                key: env.get(key) or os.environ[key]
+                for key in cls.__annotations__
+            }
+        instance._environ_from_file = cls._environ_from_file.copy()
+
+    def create_const(cls, instance: "MetaClass") -> None:
+        for key, type_hint in cls.__annotations__.items():
+            if cls.override:
+                value = instance._environ_from_file.get(key) or os.environ[key]
+            else:
+                value = os.getenv(key) or instance._environ_from_file[key]
+
+            value_typed = cls._set_type(type_hint, value)
+            setattr(instance, key, value_typed)
+            instance._environ_from_file[key] = value_typed
+
+    def _set_type(cls, type_hint: type, value: str) -> Any:
+        tmp_val: str | list[str] | map[Any] = value
+        origin: type = get_origin(type_hint) or type_hint
+        args_of_origin = get_args(type_hint)
+
+        if origin is None:
+            return None
+
+        if args_of_origin and isinstance(tmp_val, str):
+            tmp_val = tmp_val.split(",")
+
+        if (
+            origin is tuple
+            and len(args_of_origin) > 1
+            and isinstance(tmp_val, list)
+        ):
+            for i, type_ in enumerate(args_of_origin):
+                tmp_val[i] = cls._set_type(type_, tmp_val[i])
+        elif args_of_origin:
+            tmp_val = map(args_of_origin[0], tmp_val)
+
+        return origin(tmp_val)
+
+
+class Dotenv(metaclass=MetaClass):
+    def getdict(self) -> dict[str, Any]:
+        if hasattr(self, "_environ_from_file") and isinstance(
+            self._environ_from_file, dict
+        ):
+            return self._environ_from_file.copy()
+        raise TypeError("Not found _attr _environ_from_file")
+
+
+def lru_cache(
+    max_cache: int,
+) -> Callable[[Callable[F_Spec, F_Return]], Callable[F_Spec, F_Return]]:
+    cache: dict[int, EnvMap] = {}
+
+    def inner(func: Callable[F_Spec, F_Return]) -> Callable[F_Spec, F_Return]:
+        @wraps(func)
+        def wrapper(*args: F_Spec.args, **kwargs: F_Spec.kwargs) -> F_Return:
+            if len(cache) > max_cache:
+                first_key = next(iter(cache))
+                cache.pop(first_key)
+
+            key_hash = hash(f"{args}{kwargs}")
+            out_func: EnvMap = func(*args, **kwargs)
+            return cache.setdefault(key_hash, out_func)
+
+        return wrapper
+
+    return inner
+
+
+@lru_cache(max_cache=32)
+def get_from_file_env(path: str) -> EnvMap:
+    with open(path, encoding="utf-8") as file:
+        return dict(row.strip().split("=") for row in file)
diff --git a/tests/__init__.py b/tests/__init__.py
diff --git a/tests/test_dotenv.py b/tests/test_dotenv.py
@@ -0,0 +1,62 @@
+import unittest
+
+from src.envmapp.dotenv import Dotenv
+
+
+class Telegram(Dotenv, load_env="tests/.env", override=True):
+    TOKEN: str
+    SET: set[int]
+    TUPLE: tuple[str, int, str]
+    LIST: list[int]
+    FROZENSET: frozenset[int]
+
+
+class TestDotenv(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tg = Telegram()
+
+    def test_str(self) -> None:
+        self.assertIsInstance(self.tg.TOKEN, str, "tg.TOKEN is not str")
+
+    def test_set(self) -> None:
+        self.assertIsInstance(self.tg.SET, set, "tg.SET is not set")
+
+    def test_set_content(self) -> None:
+        msg = "tg.SET is not equal to content"
+        self.assertSetEqual(self.tg.SET, {312, 3213, 3215, 543, 1101}, msg)
+
+    def test_tuple(self) -> None:
+        self.assertIsInstance(self.tg.TUPLE, tuple, "tg.TUPLE is not tuple")
+
+    def test_tuple_content(self) -> None:
+        msg = "tg.TUPLE is not equal to content"
+        self.assertTupleEqual(self.tg.TUPLE, ("kaka", 10, "popa"), msg)
+
+    def test_list(self) -> None:
+        self.assertIsInstance(self.tg.LIST, list, "tg.LIST is not list")
+
+    def test_list_content(self) -> None:
+        msg = "tg.LIST is not equal to content"
+        self.assertListEqual(self.tg.LIST, [10, 11, 12, 13], msg)
+
+    def test_frozenset(self) -> None:
+        msg = "tg.FROZENSET is not frozenset"
+        self.assertIsInstance(self.tg.FROZENSET, frozenset, msg)
+
+    def test_frozenset_content(self) -> None:
+        msg = "tg.FROZENSET is not equal to content"
+        self.assertSetEqual(self.tg.FROZENSET, frozenset({1, 1, 1, 0, 1}), msg)  # noqa [B033]
+
+    def test_dict(self) -> None:
+        msg = "tg.getdict() is not dict"
+        self.assertIsInstance(self.tg.getdict(), dict, msg)
+
+    def test_dict_immutable(self) -> None:
+        msg = "tg.getdict()['TOKEN']' id is matches"
+        dct = self.tg.getdict()
+        dct["TOKEN"] = "sometoken"
+        self.assertIsNot(dct["TOKEN"], self.tg.getdict()["TOKEN"], msg)
+
+
+if __name__ == "__main__":
+    unittest.main()