"""YAML export configuration for a BusinessObjects document.
A single YAML file describes the document id, the BO filters and the output
column typing for one export. The file is loaded into a plain dictionary used
by :func:`boapi.api.export_from_config`.
Schema::
doc_id: # scalar, or a mapping by environment
dev: 123456
prod: 123456
output: out.parquet # optional, output file path
chunk_size: 10000 # optional, rows per chunk for Parquet
dataprovider: s_lieuprel # optional, dataprovider targeted by the filters
date_format: "%d/%m/%Y" # optional, default format for datetime columns
filters: # optional, BO parameters by parameter id
'0': ['2026-04-24']
'1': ['2026-05-01']
'2': [AURA, IDFR]
columns: # optional, typing and renaming
id_lieu_prel: str # shorthand: type only
lat: {type: float}
date_prel: {rename: date_prelevement, type: datetime, format: "%d/%m/%Y"}
"""
from pathlib import Path
from typing import Any, Dict, Optional
import yaml
def _resolve_doc_id(value: Any, env: Optional[str]) -> str:
"""Resolve the document id from a scalar or an environment mapping.
:param value: doc_id value from the YAML (scalar or mapping by environment)
:param env: Environment name when ``value`` is a mapping
:type env: Optional[str]
:return: Document id
:rtype: str
:raises ValueError: If the id cannot be resolved
"""
if isinstance(value, dict):
if env is None:
raise ValueError(
f"doc_id is defined per environment {list(value)}; provide env="
)
if env not in value or value[env] is None:
raise ValueError(f"No doc_id for environment '{env}' in the configuration")
return str(value[env])
if value is None:
raise ValueError("doc_id is missing from the configuration")
return str(value)
def _columns_to_mapping(columns: Dict[str, Any]) -> Dict[str, Dict[str, Any]]:
"""Convert the YAML ``columns`` section to a column mapping.
Produces the structure used by
:func:`boapi.utils.apply_column_mapping`:
``{source_name: {normalized_name, python_type, date_format}}``.
:param columns: ``columns`` section of the YAML
:type columns: Dict[str, Any]
:return: Column mapping
:rtype: Dict[str, Dict[str, Any]]
"""
mapping: Dict[str, Dict[str, Any]] = {}
for source, spec in columns.items():
if spec is None:
spec = {}
elif isinstance(spec, str):
# Shorthand: "column: type"
spec = {'type': spec}
entry: Dict[str, Any] = {
'normalized_name': spec.get('rename', source),
'python_type': spec.get('type'),
}
if spec.get('format'):
entry['date_format'] = spec['format']
mapping[source] = entry
return mapping
[docs]
def load_export_config(config_path, env: Optional[str] = None) -> Dict[str, Any]:
"""Load a YAML export configuration.
:param config_path: Path to the YAML file
:param env: Environment name used to resolve doc_id when defined per env
:type env: Optional[str]
:return: Configuration with keys doc_id, output, chunk_size, dataprovider,
date_format, filters, column_mapping
:rtype: Dict[str, Any]
:raises FileNotFoundError: If the file does not exist
"""
path = Path(config_path)
if not path.exists():
raise FileNotFoundError(f"Configuration file not found: {path}")
with open(path, 'r', encoding='utf-8') as f:
raw = yaml.safe_load(f) or {}
columns = raw.get('columns')
return {
'doc_id': _resolve_doc_id(raw.get('doc_id'), env),
'output': raw.get('output'),
'chunk_size': raw.get('chunk_size'),
'dataprovider': raw.get('dataprovider'),
'date_format': raw.get('date_format'),
'filters': raw.get('filters'),
'column_mapping': _columns_to_mapping(columns) if columns else None,
}