Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 41 additions & 11 deletions pysus/api/metadata/columns.py
Original file line number Diff line number Diff line change
Expand Up @@ -84,17 +84,35 @@ def _load_yaml_metadata(
"""Load column definitions from YAML schema files."""
result: dict[str, dict[str, Any]] = {}

# Check for database-specific schema directory
# Check for database-specific schema directory
db_dir = _SCHEMAS_DIR / database
if db_dir.exists():
for yaml_file in db_dir.glob("*.yaml"):
if group and yaml_file.stem != group:
continue
with open(yaml_file, encoding="utf-8") as f:
data = yaml.safe_load(f) or {}
for _ep_name, columns in data.items():

if group:
group_key = group.lower()

selected_items = [
(key, columns)
for key, columns in data.items()
if key.lower() == group_key
or key.lower().startswith(f"{group_key}_")
]

if not selected_items and yaml_file.stem.lower() == group_key:
selected_items = list(data.items())

if not selected_items:
continue
else:
selected_items = list(data.items())

for _ep_name, columns in selected_items:
if not isinstance(columns, list):
continue

for col_def in columns:
col_name = col_def.get("name", "").upper()
if col_name:
Expand Down Expand Up @@ -195,12 +213,24 @@ def available_groups(database: str) -> list[str]:

db_dir = _SCHEMAS_DIR / database
if db_dir.exists():
groups.extend(p.stem for p in db_dir.glob("*.yaml"))
for yaml_file in db_dir.glob("*.yaml"):
groups.append(yaml_file.stem)

with open(yaml_file, encoding="utf-8") as f:
data = yaml.safe_load(f) or {}

for key in data:
if isinstance(key, str):
group = key.split("_", 1)[0]
if group not in groups:
groups.append(group)

if database == "sinan":
saude_dir = Path(__file__).parent.parent / "saude" / "schemas"
if saude_dir.exists():
for yaml_file in saude_dir.glob("*.yaml"):
if yaml_file.stem not in groups:
groups.append(yaml_file.stem)

saude_dir = Path(__file__).parent.parent / "saude" / "schemas"
if saude_dir.exists():
for yaml_file in saude_dir.glob("*.yaml"):
if yaml_file.stem not in groups:
groups.append(yaml_file.stem)

return sorted(groups)
return sorted(groups)
24 changes: 24 additions & 0 deletions pysus/tests/api/test_metadata_columns.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,14 @@ def test_load_sia_metadata(self):
assert isinstance(meta, dict)
assert len(meta) > 0

def test_load_sia_bi_metadata(self):
"""Test loading SIA BI metadata."""
meta = load_column_metadata("sia", group="bi")
assert isinstance(meta, dict)
assert len(meta) > 0
assert "CNES" in meta
assert "PROC_REA" in meta

def test_load_sim_metadata(self):
"""Test loading SIM metadata."""
meta = load_column_metadata("sim", group="do")
Expand Down Expand Up @@ -109,3 +117,19 @@ def test_sinan_has_groups(self):
"""Test that SINAN has groups."""
groups = available_groups("sinan")
assert len(groups) > 0

def test_detects_groups_from_yaml_keys(self):
"""Test that groups are detected from YAML root keys."""
groups = available_groups("sia")

assert "pa" in groups
assert "bi" in groups

def test_non_sinan_excludes_saude_groups(self):
"""Test that saude schema groups are only included for SINAN."""
groups = available_groups("sia")

assert "bi" in groups
assert "pa" in groups
assert "arboviroses" not in groups
assert "vigilanciameioambiente" not in groups
Loading