feat(infra): funciones Python config_from_env y dotenv_load con tests
- config_from_env_py_infra: dataclass + field metadata, tipos str/int/float/bool/list - dotenv_load_py_infra: parser .env con semantica de no-sobreescritura - 15 tests unitarios Python, todos PASS - registry.db actualizado con fn index (685 funciones, 109 tipos) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,54 @@
|
||||
---
|
||||
name: config_from_env
|
||||
kind: function
|
||||
lang: py
|
||||
domain: infra
|
||||
version: "1.0.0"
|
||||
purity: impure
|
||||
signature: "config_from_env(target_class: type) -> object"
|
||||
description: "Crea una instancia de un dataclass poblada desde variables de entorno usando field metadata (env, required, default, secret). Soporta str, int, float, bool, list. Acumula todos los errores antes de lanzar ValueError."
|
||||
tags: [config, env, dataclass, infra, python]
|
||||
uses_functions: []
|
||||
uses_types: []
|
||||
returns: []
|
||||
returns_optional: false
|
||||
error_type: "error_go_core"
|
||||
imports: [os, dataclasses]
|
||||
params:
|
||||
- name: target_class
|
||||
desc: "clase dataclass con field metadata declarando env, required, default y/o secret por cada campo"
|
||||
output: "instancia del dataclass con todos los campos poblados desde env o defaults"
|
||||
tested: true
|
||||
tests:
|
||||
- "populate string desde env"
|
||||
- "populate int con conversion"
|
||||
- "populate bool true"
|
||||
- "populate float field"
|
||||
- "populate list comma separated"
|
||||
- "default usado cuando env no seteada"
|
||||
- "required sin valor lanza error"
|
||||
- "no dataclass lanza type error"
|
||||
test_file_path: "python/functions/infra/config_from_env_test.py"
|
||||
file_path: "python/functions/infra/config_from_env.py"
|
||||
---
|
||||
|
||||
## Ejemplo
|
||||
|
||||
```python
|
||||
from dataclasses import dataclass, field
|
||||
from config_from_env import config_from_env
|
||||
|
||||
@dataclass
|
||||
class AppConfig:
|
||||
host: str = field(default="localhost", metadata={"env": "HOST", "default": "localhost"})
|
||||
port: int = field(default=8080, metadata={"env": "PORT", "default": "8080", "required": True})
|
||||
api_key: str = field(default="", metadata={"env": "API_KEY", "required": True, "secret": True})
|
||||
tags: list = field(default_factory=list, metadata={"env": "TAGS", "default": "prod,stable"})
|
||||
|
||||
cfg = config_from_env(AppConfig)
|
||||
print(cfg.port) # 8080 o el valor de PORT
|
||||
```
|
||||
|
||||
## Notas
|
||||
|
||||
Solo stdlib (os, dataclasses). Conversion de tipos: str sin cambio, int/float con parseo, bool desde "1"/"true"/"yes", list desde split por coma. El tag secret no tiene efecto en runtime. La funcion inspecciona todos los fields via dataclasses.fields() y solo procesa los que tienen metadata["env"].
|
||||
@@ -0,0 +1,106 @@
|
||||
"""config_from_env — popula dataclasses desde variables de entorno usando field metadata."""
|
||||
|
||||
import os
|
||||
from dataclasses import fields as dc_fields
|
||||
|
||||
|
||||
def config_from_env(target_class: type) -> object:
|
||||
"""Crea una instancia de un dataclass poblada desde variables de entorno.
|
||||
|
||||
Cada campo del dataclass puede tener metadata que guia el mapeo:
|
||||
- ``env``: nombre de la variable de entorno (obligatorio para usar este sistema)
|
||||
- ``required``: bool, si True y la var no esta seteada lanza ValueError
|
||||
- ``default``: valor string usado cuando la var no esta seteada (opcional)
|
||||
- ``secret``: bool, documentacion solamente, sin efecto en runtime
|
||||
|
||||
La conversion de tipo es automatica segun el tipo anotado del campo:
|
||||
- ``str``: sin conversion
|
||||
- ``int``: int(value)
|
||||
- ``float``: float(value)
|
||||
- ``bool``: True para "1", "true", "yes" (case-insensitive); False para el resto
|
||||
- ``list[str]`` o ``list``: split por coma, stripping espacios
|
||||
|
||||
Args:
|
||||
target_class: clase dataclass con campo metadata en sus fields.
|
||||
|
||||
Returns:
|
||||
instancia de target_class con todos los campos poblados.
|
||||
|
||||
Raises:
|
||||
ValueError: si un campo requerido no esta seteado, o si la conversion de tipo falla.
|
||||
TypeError: si target_class no es un dataclass.
|
||||
"""
|
||||
try:
|
||||
all_fields = dc_fields(target_class) # raises TypeError if not dataclass
|
||||
except TypeError:
|
||||
raise TypeError(f"config_from_env: {target_class.__name__} is not a dataclass")
|
||||
|
||||
kwargs: dict = {}
|
||||
errors: list[str] = []
|
||||
|
||||
for f in all_fields:
|
||||
meta = f.metadata
|
||||
env_key: str | None = meta.get("env")
|
||||
if env_key is None:
|
||||
# no env tag → leave at dataclass default (will be provided via default_factory)
|
||||
continue
|
||||
|
||||
required: bool = meta.get("required", False)
|
||||
default_str: str | None = meta.get("default")
|
||||
|
||||
raw = os.environ.get(env_key, "")
|
||||
if not raw:
|
||||
if default_str is not None:
|
||||
raw = default_str
|
||||
elif required:
|
||||
errors.append(f"field '{f.name}': env var '{env_key}' is required but not set")
|
||||
continue
|
||||
else:
|
||||
continue # optional, unset — use dataclass default
|
||||
|
||||
# type conversion
|
||||
try:
|
||||
kwargs[f.name] = _coerce(raw, f.type, f.name)
|
||||
except (ValueError, TypeError) as exc:
|
||||
errors.append(str(exc))
|
||||
|
||||
if errors:
|
||||
raise ValueError("config_from_env: " + "; ".join(errors))
|
||||
|
||||
return target_class(**kwargs)
|
||||
|
||||
|
||||
def _coerce(raw: str, annotation: type | str, field_name: str) -> object:
|
||||
"""Coerce raw string to the annotated type."""
|
||||
# handle string annotations (from __future__ import annotations or forward refs)
|
||||
if isinstance(annotation, str):
|
||||
ann_str = annotation.lower().strip()
|
||||
else:
|
||||
# Get the type name for comparison
|
||||
ann_str = getattr(annotation, "__name__", str(annotation)).lower()
|
||||
|
||||
if ann_str in ("str", "string"):
|
||||
return raw
|
||||
|
||||
if ann_str == "int":
|
||||
try:
|
||||
return int(raw)
|
||||
except ValueError:
|
||||
raise ValueError(f"field '{field_name}': cannot convert {raw!r} to int")
|
||||
|
||||
if ann_str == "float":
|
||||
try:
|
||||
return float(raw)
|
||||
except ValueError:
|
||||
raise ValueError(f"field '{field_name}': cannot convert {raw!r} to float")
|
||||
|
||||
if ann_str == "bool":
|
||||
return raw.strip().lower() in ("1", "true", "yes")
|
||||
|
||||
# list / list[str]
|
||||
if "list" in ann_str:
|
||||
parts = [p.strip() for p in raw.split(",") if p.strip()]
|
||||
return parts
|
||||
|
||||
# fallback: return as string
|
||||
return raw
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Tests para config_from_env."""
|
||||
|
||||
import os
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
import pytest
|
||||
|
||||
from config_from_env import config_from_env
|
||||
|
||||
|
||||
@dataclass
|
||||
class SimpleConfig:
|
||||
host: str = field(default="localhost", metadata={"env": "CFE_PY_HOST", "default": "localhost"})
|
||||
port: int = field(default=8080, metadata={"env": "CFE_PY_PORT", "default": "8080"})
|
||||
debug: bool = field(default=False, metadata={"env": "CFE_PY_DEBUG", "default": "false"})
|
||||
ratio: float = field(default=1.0, metadata={"env": "CFE_PY_RATIO", "default": "1.0"})
|
||||
tags: list = field(default_factory=list, metadata={"env": "CFE_PY_TAGS", "default": "a,b"})
|
||||
|
||||
|
||||
@dataclass
|
||||
class RequiredConfig:
|
||||
api_key: str = field(default="", metadata={"env": "CFE_PY_API_KEY", "required": True})
|
||||
|
||||
|
||||
def _clean(*keys: str) -> None:
|
||||
for k in keys:
|
||||
os.environ.pop(k, None)
|
||||
|
||||
|
||||
def test_populate_string_desde_env():
|
||||
os.environ["CFE_PY_HOST"] = "myhost"
|
||||
try:
|
||||
cfg = config_from_env(SimpleConfig)
|
||||
assert cfg.host == "myhost"
|
||||
finally:
|
||||
_clean("CFE_PY_HOST")
|
||||
|
||||
|
||||
def test_populate_int_con_conversion():
|
||||
os.environ["CFE_PY_PORT"] = "9090"
|
||||
try:
|
||||
cfg = config_from_env(SimpleConfig)
|
||||
assert cfg.port == 9090
|
||||
finally:
|
||||
_clean("CFE_PY_PORT")
|
||||
|
||||
|
||||
def test_populate_bool_true():
|
||||
os.environ["CFE_PY_DEBUG"] = "true"
|
||||
try:
|
||||
cfg = config_from_env(SimpleConfig)
|
||||
assert cfg.debug is True
|
||||
finally:
|
||||
_clean("CFE_PY_DEBUG")
|
||||
|
||||
|
||||
def test_populate_float_field():
|
||||
os.environ["CFE_PY_RATIO"] = "3.14"
|
||||
try:
|
||||
cfg = config_from_env(SimpleConfig)
|
||||
assert abs(cfg.ratio - 3.14) < 1e-9
|
||||
finally:
|
||||
_clean("CFE_PY_RATIO")
|
||||
|
||||
|
||||
def test_populate_list_comma_separated():
|
||||
os.environ["CFE_PY_TAGS"] = "x,y,z"
|
||||
try:
|
||||
cfg = config_from_env(SimpleConfig)
|
||||
assert cfg.tags == ["x", "y", "z"]
|
||||
finally:
|
||||
_clean("CFE_PY_TAGS")
|
||||
|
||||
|
||||
def test_default_usado_cuando_env_no_seteada():
|
||||
_clean("CFE_PY_PORT")
|
||||
cfg = config_from_env(SimpleConfig)
|
||||
assert cfg.port == 8080
|
||||
|
||||
|
||||
def test_required_sin_valor_lanza_error():
|
||||
_clean("CFE_PY_API_KEY")
|
||||
with pytest.raises(ValueError, match="required"):
|
||||
config_from_env(RequiredConfig)
|
||||
|
||||
|
||||
def test_no_dataclass_lanza_type_error():
|
||||
class NotADataclass:
|
||||
pass
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
config_from_env(NotADataclass)
|
||||
@@ -0,0 +1,46 @@
|
||||
---
|
||||
name: dotenv_load
|
||||
kind: function
|
||||
lang: py
|
||||
domain: infra
|
||||
version: "1.0.0"
|
||||
purity: impure
|
||||
signature: "dotenv_load(path: str) -> None"
|
||||
description: "Parsea un archivo .env (KEY=VALUE) e inyecta pares en os.environ. Ignora comentarios # y lineas vacias. No sobreescribe variables ya presentes. Soporta valores con comillas simples o dobles."
|
||||
tags: [dotenv, env, config, os, infra, python]
|
||||
uses_functions: []
|
||||
uses_types: []
|
||||
returns: []
|
||||
returns_optional: false
|
||||
error_type: "error_go_core"
|
||||
imports: [os]
|
||||
params:
|
||||
- name: path
|
||||
desc: "ruta al archivo .env a parsear (ej: '.env', 'config/.env.production')"
|
||||
output: "None — efecto lateral: variables seteadas en os.environ"
|
||||
tested: true
|
||||
tests:
|
||||
- "parsea key value y setea variable"
|
||||
- "ignora comentarios y lineas vacias"
|
||||
- "no sobreescribe variables existentes"
|
||||
- "valores con comillas dobles se stripean"
|
||||
- "valores con comillas simples se stripean"
|
||||
- "archivo inexistente lanza error"
|
||||
- "linea sin igual lanza value error"
|
||||
test_file_path: "python/functions/infra/dotenv_load_test.py"
|
||||
file_path: "python/functions/infra/dotenv_load.py"
|
||||
---
|
||||
|
||||
## Ejemplo
|
||||
|
||||
```python
|
||||
from dotenv_load import dotenv_load
|
||||
import os
|
||||
|
||||
dotenv_load(".env")
|
||||
db_url = os.environ["DATABASE_URL"]
|
||||
```
|
||||
|
||||
## Notas
|
||||
|
||||
Solo stdlib (os). Semantica de no-sobreescritura: variables inyectadas por Docker/CI/CD tienen precedencia sobre el .env local. Separador es el primer '=' encontrado en la linea (permite valores con '='). Formato: KEY=VALUE, KEY="val", KEY='val'.
|
||||
@@ -0,0 +1,53 @@
|
||||
"""dotenv_load — parsea un archivo .env y popula os.environ sin sobreescribir."""
|
||||
|
||||
import os
|
||||
|
||||
|
||||
def dotenv_load(path: str) -> None:
|
||||
"""Parsea un archivo .env y setea variables en os.environ.
|
||||
|
||||
Reglas de parseo:
|
||||
- Lineas que empiezan con # (tras strip) son comentarios y se ignoran.
|
||||
- Lineas vacias se ignoran.
|
||||
- Formato esperado: KEY=VALUE (separador es el primer =).
|
||||
- Valores con comillas simples o dobles externas se stripean.
|
||||
- Variables ya presentes en os.environ NO se sobreescriben.
|
||||
|
||||
Args:
|
||||
path: ruta al archivo .env.
|
||||
|
||||
Returns:
|
||||
None
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: si el archivo no existe.
|
||||
ValueError: si una linea no tiene el separador '='.
|
||||
"""
|
||||
with open(path, encoding="utf-8") as fh:
|
||||
for lineno, raw_line in enumerate(fh, start=1):
|
||||
line = raw_line.strip()
|
||||
|
||||
# skip blanks and comments
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
|
||||
if "=" not in line:
|
||||
raise ValueError(
|
||||
f"dotenv_load: {path!r} line {lineno}: missing '=' in {line!r}"
|
||||
)
|
||||
|
||||
key, _, val = line.partition("=")
|
||||
key = key.strip()
|
||||
val = _strip_quotes(val)
|
||||
|
||||
# only set if not already present
|
||||
if key not in os.environ or os.environ[key] == "":
|
||||
os.environ[key] = val
|
||||
|
||||
|
||||
def _strip_quotes(s: str) -> str:
|
||||
"""Remove surrounding single or double quotes."""
|
||||
if len(s) >= 2:
|
||||
if (s[0] == '"' and s[-1] == '"') or (s[0] == "'" and s[-1] == "'"):
|
||||
return s[1:-1]
|
||||
return s
|
||||
@@ -0,0 +1,91 @@
|
||||
"""Tests para dotenv_load."""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
|
||||
from dotenv_load import dotenv_load
|
||||
|
||||
|
||||
def write_env(content: str) -> str:
|
||||
"""Write content to a temp file and return its path."""
|
||||
fd, path = tempfile.mkstemp(suffix=".env")
|
||||
os.write(fd, content.encode())
|
||||
os.close(fd)
|
||||
return path
|
||||
|
||||
|
||||
def _clean(*keys: str) -> None:
|
||||
for k in keys:
|
||||
os.environ.pop(k, None)
|
||||
|
||||
|
||||
def test_parsea_key_value_y_setea_variable():
|
||||
path = write_env("DL_PY_FOO=hello\n")
|
||||
_clean("DL_PY_FOO")
|
||||
try:
|
||||
dotenv_load(path)
|
||||
assert os.environ.get("DL_PY_FOO") == "hello"
|
||||
finally:
|
||||
_clean("DL_PY_FOO")
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
def test_ignora_comentarios_y_lineas_vacias():
|
||||
content = "# comment\n\nDL_PY_BAR=world\n"
|
||||
path = write_env(content)
|
||||
_clean("DL_PY_BAR")
|
||||
try:
|
||||
dotenv_load(path)
|
||||
assert os.environ.get("DL_PY_BAR") == "world"
|
||||
finally:
|
||||
_clean("DL_PY_BAR")
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
def test_no_sobreescribe_variables_existentes():
|
||||
path = write_env("DL_PY_EXIST=new\n")
|
||||
os.environ["DL_PY_EXIST"] = "original"
|
||||
try:
|
||||
dotenv_load(path)
|
||||
assert os.environ["DL_PY_EXIST"] == "original"
|
||||
finally:
|
||||
_clean("DL_PY_EXIST")
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
def test_valores_con_comillas_dobles_se_stripean():
|
||||
path = write_env('DL_PY_QUOTED="quoted value"\n')
|
||||
_clean("DL_PY_QUOTED")
|
||||
try:
|
||||
dotenv_load(path)
|
||||
assert os.environ.get("DL_PY_QUOTED") == "quoted value"
|
||||
finally:
|
||||
_clean("DL_PY_QUOTED")
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
def test_valores_con_comillas_simples_se_stripean():
|
||||
path = write_env("DL_PY_SINGLE='single quoted'\n")
|
||||
_clean("DL_PY_SINGLE")
|
||||
try:
|
||||
dotenv_load(path)
|
||||
assert os.environ.get("DL_PY_SINGLE") == "single quoted"
|
||||
finally:
|
||||
_clean("DL_PY_SINGLE")
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
def test_archivo_inexistente_lanza_error():
|
||||
with pytest.raises(FileNotFoundError):
|
||||
dotenv_load("/nonexistent/.env.does.not.exist")
|
||||
|
||||
|
||||
def test_linea_sin_igual_lanza_value_error():
|
||||
path = write_env("INVALID_LINE\n")
|
||||
try:
|
||||
with pytest.raises(ValueError, match="missing '='"):
|
||||
dotenv_load(path)
|
||||
finally:
|
||||
os.unlink(path)
|
||||
Reference in New Issue
Block a user