Merge pull request #4 from whacked/update-custom-dotenv-path

update handling of non-default .env paths

Author: tutankalex

flake.nix

@@ -80,6 +80,7 @@
         '' + ''
           ### SHORTCUTS
           alias check='python -m schematized_config'
+          alias start-jupyter='jupyter notebook'
         '';  # join strings with +
       };
 

nbs/00_core.ipynb

@@ -1,6 +1,7 @@
 {
  "cells": [
   {
+   "attachments": {},
    "cell_type": "markdown",
    "metadata": {},
    "source": [
@@ -221,23 +222,25 @@
     "    CONFIG_VALIDATOR_JSON_SCHEMA_ENVVAR_NAME = 'CONFIG_VALIDATOR_JSON_SCHEMA'\n",
     "\n",
     "    @classmethod\n",
-    "    def load_json(cls, json_source: Union[str, dict]=None) -> dict:\n",
+    "    def load_json(cls, json_source: Union[str, dict]=None, storage_driver: FS = None) -> dict:\n",
     "        '''\n",
     "        convenience method to return a dict from either\n",
     "        a file path or an already-loaded dict\n",
     "        '''\n",
+    "        storage_driver = storage_driver or cls.DEFAULT_STORAGE_DRIVER\n",
     "        if isinstance(json_source, str):\n",
-    "            with cls.DEFAULT_STORAGE_DRIVER.open(json_source) as ifile:\n",
+    "            with storage_driver.open(json_source) as ifile:\n",
     "                return json.load(ifile)\n",
     "        elif isinstance(json_source, dict):\n",
     "            return json_source\n",
     "\n",
     "    @classmethod\n",
-    "    def get_default_json_schema(cls):\n",
+    "    def get_default_json_schema(cls, storage_driver: FS = None) -> dict:\n",
+    "        storage_driver = storage_driver or cls.DEFAULT_STORAGE_DRIVER\n",
     "        if cls.CONFIG_VALIDATOR_JSON_SCHEMA_ENVVAR_NAME in os.environ:\n",
     "            expected_json_schema_path = \\\n",
     "                os.environ[cls.CONFIG_VALIDATOR_JSON_SCHEMA_ENVVAR_NAME]\n",
-    "            with cls.DEFAULT_STORAGE_DRIVER.open(expected_json_schema_path) as ifile:\n",
+    "            with storage_driver.open(expected_json_schema_path) as ifile:\n",
     "                return json.load(ifile)\n",
     "        return None\n",
     "\n",
@@ -270,16 +273,18 @@
     "        return coerced_config\n",
     "    \n",
     "    @classmethod\n",
-    "    def load_validated_config(cls, json_schema: Union[str, dict], config: dict):\n",
-    "        return cls(json_schema).load_config(config)\n",
+    "    def load_validated_config(cls, json_schema: Union[str, dict], config: dict, **kwargs):\n",
+    "        return cls(json_schema, **kwargs).load_config(config)\n",
     "\n",
     "    @classmethod\n",
-    "    def load_validated_environment(cls, json_schema: Union[str, dict]=None):\n",
-    "        return cls.load_validated_config(json_schema, dict(os.environ))\n",
+    "    def load_validated_environment(cls, json_schema: Union[str, dict]=None, **kwargs):\n",
+    "        return cls.load_validated_config(json_schema, dict(os.environ), **kwargs)\n",
     "        \n",
     "    @classmethod\n",
-    "    def load_dotenv(cls, json_schema: Union[str, dict]=None, dotenv_path: str=None):\n",
-    "        config = dotenv.dotenv_values(dotenv_path)\n",
+    "    def load_dotenv(cls, json_schema: Union[str, dict]=None, dotenv_path: str=None, storage_driver: FS=None):\n",
+    "        storage_driver = storage_driver or cls.DEFAULT_STORAGE_DRIVER\n",
+    "        with storage_driver.open(dotenv_path) as ifile:\n",
+    "            config = dotenv.dotenv_values(stream=ifile)\n",
     "        return cls.load_validated_config(\n",
     "            json_schema or cls.get_default_json_schema(), config)"
    ]
@@ -509,7 +514,8 @@
     "from fs.memoryfs import MemoryFS\n",
     "memfs = MemoryFS()\n",
     "\n",
-    "with memfs.open('schema.json', 'w') as ofile:\n",
+    "memfs.makedirs('extra-long-directory-place', recreate=True)\n",
+    "with memfs.open('extra-long-directory-place/schema.json', 'w') as ofile:\n",
     "    ofile.write(json.dumps(example_properties_schema))\n",
     "    os.environ['CONFIG_VALIDATOR_JSON_SCHEMA'] = ofile.name\n",
     "\n",
@@ -526,7 +532,26 @@
     "    'MY_INTEGER_VALUE': -85,\n",
     "    'A_NUMERIC_VALUE': 12300.0,\n",
     "    '_____A_STRING_VALUE____with_default__': 'underscores_and spaces',\n",
-    "})"
+    "})\n",
+    "\n",
+    "# test loading dotenv from an arbitrary file\n",
+    "memfs.makedirs('special-bespoke-location', recreate=True)\n",
+    "with memfs.open('special-bespoke-location/my-own.env', 'w') as ofile:\n",
+    "    ofile.write('\\n'.join([\n",
+    "        'string_value_with_enum=only',\n",
+    "        'MY_INTEGER_VALUE=9989998',\n",
+    "        'A_NUMERIC_VALUE=1167.89',\n",
+    "    ]))\n",
+    "\n",
+    "validated_dotenv = validator.load_dotenv(dotenv_path='special-bespoke-location/my-own.env')\n",
+    "test_eq(validated_dotenv, {\n",
+    "    'string_value_with_enum': 'only',\n",
+    "    'MY_INTEGER_VALUE': 9989998,\n",
+    "    'A_NUMERIC_VALUE': 1167.89,\n",
+    "    '_____A_STRING_VALUE____with_default__': 'underscores_and spaces',\n",
+    "})\n",
+    "\n",
+    "test_fail(validator.load_dotenv, kwargs={'dotenv_path': 'non-existent-location-own.env'})"
    ]
   },
   {

nbs/index.ipynb

@@ -35,6 +35,22 @@
     "```"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# development"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "assuming you have [nix](https://nixos.org/download.html) installed and ready, make sure `nix-command` and `flake` are enabled (oneliner: run `export NIX_CONFIG=\"experimental-features = nix-command flakes\"` in the terminal), then enter the dev shell using `nix develop`\n",
+    "\n",
+    "start the jupyter notebook using the provided alias or just `jupyter notebook`, and hack away"
+   ]
+  },
   {
    "cell_type": "markdown",
    "metadata": {},

schematized_config/core.py

@@ -98,23 +98,25 @@ class ConfigValidator(object):
     CONFIG_VALIDATOR_JSON_SCHEMA_ENVVAR_NAME = 'CONFIG_VALIDATOR_JSON_SCHEMA'
 
     @classmethod
-    def load_json(cls, json_source: Union[str, dict]=None) -> dict:
+    def load_json(cls, json_source: Union[str, dict]=None, storage_driver: FS = None) -> dict:
         '''
         convenience method to return a dict from either
         a file path or an already-loaded dict
         '''
+        storage_driver = storage_driver or cls.DEFAULT_STORAGE_DRIVER
         if isinstance(json_source, str):
-            with cls.DEFAULT_STORAGE_DRIVER.open(json_source) as ifile:
+            with storage_driver.open(json_source) as ifile:
                 return json.load(ifile)
         elif isinstance(json_source, dict):
             return json_source
 
     @classmethod
-    def get_default_json_schema(cls):
+    def get_default_json_schema(cls, storage_driver: FS = None) -> dict:
+        storage_driver = storage_driver or cls.DEFAULT_STORAGE_DRIVER
         if cls.CONFIG_VALIDATOR_JSON_SCHEMA_ENVVAR_NAME in os.environ:
             expected_json_schema_path = \
                 os.environ[cls.CONFIG_VALIDATOR_JSON_SCHEMA_ENVVAR_NAME]
-            with cls.DEFAULT_STORAGE_DRIVER.open(expected_json_schema_path) as ifile:
+            with storage_driver.open(expected_json_schema_path) as ifile:
                 return json.load(ifile)
         return None
 
@@ -147,15 +149,17 @@ def load_config(self, config: dict):
         return coerced_config
 
     @classmethod
-    def load_validated_config(cls, json_schema: Union[str, dict], config: dict):
-        return cls(json_schema).load_config(config)
+    def load_validated_config(cls, json_schema: Union[str, dict], config: dict, **kwargs):
+        return cls(json_schema, **kwargs).load_config(config)
 
     @classmethod
-    def load_validated_environment(cls, json_schema: Union[str, dict]=None):
-        return cls.load_validated_config(json_schema, dict(os.environ))
+    def load_validated_environment(cls, json_schema: Union[str, dict]=None, **kwargs):
+        return cls.load_validated_config(json_schema, dict(os.environ), **kwargs)
 
     @classmethod
-    def load_dotenv(cls, json_schema: Union[str, dict]=None, dotenv_path: str=None):
-        config = dotenv.dotenv_values(dotenv_path)
+    def load_dotenv(cls, json_schema: Union[str, dict]=None, dotenv_path: str=None, storage_driver: FS=None):
+        storage_driver = storage_driver or cls.DEFAULT_STORAGE_DRIVER
+        with storage_driver.open(dotenv_path) as ifile:
+            config = dotenv.dotenv_values(stream=ifile)
         return cls.load_validated_config(
             json_schema or cls.get_default_json_schema(), config)