Base de Datos

Database configuration helpers.

database.config.get_database_url()

Return the configured database URL.

Devuelve:

Database URL from DATABASE_URL or the local SQLite fallback.

Tipo del valor devuelto:

str

database.config.is_sqlite_url(database_url)

Check whether a database URL targets SQLite.

Parámetros:

database_url (str) – SQLAlchemy database URL.

Devuelve:

True when the URL uses the SQLite dialect.

Tipo del valor devuelto:

bool

Command-line initializer for the IGSM database.

database.init_db.init_database(include_demo_results=False)

Create the configured database schema and seed reference data.

Parámetros:

include_demo_results (bool) – Deprecated compatibility flag. Demo app tables are no longer created by the ORM.

Tipo del valor devuelto:

None

database.init_db.main()

Run the database initializer CLI.

Tipo del valor devuelto:

None

SQLAlchemy models for the IGSM dimensional and fact data model.

database.models.utcnow()

Return the current timezone-aware UTC timestamp.

Devuelve:

Current UTC datetime with timezone information.

Tipo del valor devuelto:

datetime

class database.models.Base(*args, **kwargs)

Bases: DeclarativeBase

Base class for all IGSM ORM models.

Parámetros:

  • args (Any)

  • kwargs (Any)

Tipo del valor devuelto:

Any

class database.models.DMMunicipality(*args, **kwargs)

Bases: Base

Municipality dimension row.

Variables:

  • municipality_id (Mapped[int]) – Primary key.

  • code (Mapped[str]) – Official municipality code.

  • name (Mapped[str]) – Municipality display name.

  • province (Mapped[str | None]) – Province name.

  • region (Mapped[str | None]) – Planning region name.

  • latitude (Mapped[float | None]) – Latitude used for map visualizations.

  • longitude (Mapped[float | None]) – Longitude used for map visualizations.

  • diversified_services (Mapped[list['DMMunicipalityDiversifiedService']]) – Diversified-service links for the municipality.

  • responses (Mapped[list['FactIndicatorResponse']]) – Indicator response fact rows submitted by the municipality.

Parámetros:

  • args (Any)

  • kwargs (Any)

Tipo del valor devuelto:

Any

class database.models.DMMunicipalityDiversifiedService(*args, **kwargs)

Bases: Base

Diversified-service link for a municipality.

Variables:

  • municipality_id (Mapped[int]) – Municipality primary key.

  • service_id (Mapped[int]) – Service primary key.

  • municipality (Mapped[DMMunicipality]) – Linked municipality row.

  • service (Mapped['DMService']) – Linked service row.

Parámetros:

  • args (Any)

  • kwargs (Any)

Tipo del valor devuelto:

Any

class database.models.DMAxis(*args, **kwargs)

Bases: Base

IGSM management axis dimension row.

Variables:

  • axis_id (Mapped[int]) – Primary key.

  • name (Mapped[str]) – Axis display name.

  • services (Mapped[list['DMService']]) – Services assigned to the axis.

Parámetros:

  • args (Any)

  • kwargs (Any)

Tipo del valor devuelto:

Any

class database.models.DMService(*args, **kwargs)

Bases: Base

IGSM service dimension row.

Variables:

  • service_id (Mapped[int]) – Primary key.

  • axis_id (Mapped[int]) – Parent axis primary key.

  • name (Mapped[str]) – Service display name.

  • service_code (Mapped[str]) – Official service code.

  • grouping (Mapped[str | None]) – Service grouping, such as basic or diversified.

  • diversified_key (Mapped[str | None]) – Optional diversified-service key used by municipalities.

  • axis (Mapped[DMAxis]) – Parent axis row.

  • indicators (Mapped[list['DMIndicator']]) – Indicators assigned to the service.

  • municipality_links (Mapped[list[DMMunicipalityDiversifiedService]]) – Municipality links for diversified services.

Parámetros:

  • args (Any)

  • kwargs (Any)

Tipo del valor devuelto:

Any

class database.models.DMStage(*args, **kwargs)

Bases: Base

Global IGSM stage dimension row.

Variables:

  • stage_id (Mapped[int]) – Primary key.

  • name (Mapped[str]) – Stage display name.

  • indicators (Mapped[list['DMIndicator']]) – Indicators assigned to the stage.

Parámetros:

  • args (Any)

  • kwargs (Any)

Tipo del valor devuelto:

Any

class database.models.DMIndicator(*args, **kwargs)

Bases: Base

IGSM indicator dimension row.

Variables:

  • indicator_id (Mapped[int]) – Primary key.

  • service_id (Mapped[int]) – Parent service primary key.

  • stage_id (Mapped[int]) – Parent stage primary key.

  • code (Mapped[str]) – Official indicator code.

  • name (Mapped[str]) – Indicator display name.

  • type (Mapped[str | None]) – Indicator response type.

  • evidence_required (Mapped[bool | None]) – Whether the indicator requires supporting evidence.

  • documentation (Mapped[str | None]) – Evidence or documentation guidance.

  • service (Mapped[DMService]) – Parent service row.

  • stage (Mapped[DMStage]) – Parent stage row.

  • responses (Mapped[list['FactIndicatorResponse']]) – Response fact rows for the indicator.

Parámetros:

  • args (Any)

  • kwargs (Any)

Tipo del valor devuelto:

Any

class database.models.FactIndicatorResponse(*args, **kwargs)

Bases: Base

Numeric indicator answer submitted by a municipality.

Variables:

  • response_id (Mapped[int]) – Primary key.

  • date_time (Mapped[datetime]) – Submission timestamp.

  • municipality_id (Mapped[int]) – Municipality primary key.

  • indicator_id (Mapped[int]) – Indicator primary key.

  • value (Mapped[float]) – Numeric answer value.

  • municipality (Mapped[DMMunicipality]) – Linked municipality row.

  • indicator (Mapped[DMIndicator]) – Linked indicator row.

Parámetros:

  • args (Any)

  • kwargs (Any)

Tipo del valor devuelto:

Any

class database.models.FactStageWeight(*args, **kwargs)

Bases: Base

Effective-dated global stage weights.

Variables:

  • stage_weight_id (Mapped[int]) – Primary key.

  • planning_weight (Mapped[float]) – Weight for the planning stage.

  • execution_weight (Mapped[float]) – Weight for the execution stage.

  • evaluation_weight (Mapped[float]) – Weight for the evaluation stage.

  • effective_from (Mapped[date]) – First date when the weights apply.

Parámetros:

  • args (Any)

  • kwargs (Any)

Tipo del valor devuelto:

Any

class database.models.FactMaturityThreshold(*args, **kwargs)

Bases: Base

Effective-dated maturity-level thresholds.

Variables:

  • threshold_id (Mapped[int]) – Primary key.

  • initial_upper (Mapped[float]) – Upper bound for the initial level.

  • basic_upper (Mapped[float]) – Upper bound for the basic level.

  • intermediate_upper (Mapped[float]) – Upper bound for the intermediate level.

  • advanced_upper (Mapped[float]) – Upper bound for the advanced level.

  • optimizing_upper (Mapped[float]) – Upper bound for the optimizing level.

  • effective_from (Mapped[date]) – First date when the thresholds apply.

Parámetros:

  • args (Any)

  • kwargs (Any)

Tipo del valor devuelto:

Any

Repository functions for the IGSM ORM package.

database.repositories._managed_session(session=None)

Return a caller-owned session or create a managed session.

Parámetros:

session (Session | None) – Optional SQLAlchemy session supplied by the caller.

Devuelve:

Context manager that yields a SQLAlchemy session.

database.repositories._one_or_none(session, statement)

Execute a scalar statement and return its first row.

Parámetros:

  • session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

  • statement (sqlalchemy.Select) – SQLAlchemy select statement.

Devuelve:

First scalar result, or None when no row matches.

database.repositories._parse_end_date(end_date)

Parse an end-date value.

Parámetros:

end_date (date | datetime | str | None) – Optional cutoff date as a date, datetime, YYYY-MM-DD, or legacy YYYY string.

Devuelve:

Parsed date or datetime. A legacy year string maps to December 31 of that year.

Muestra:

ValueError – If the value cannot be parsed as a date.

Tipo del valor devuelto:

date | datetime | None

database.repositories._end_datetime(end_date)

Build an inclusive datetime cutoff for an end date.

Parámetros:

end_date (date | datetime | str | None) – Optional cutoff date.

Devuelve:

Datetime cutoff at the end of the requested day.

Tipo del valor devuelto:

datetime

database.repositories._timestamp_for_end_date(end_date)

Build a response timestamp for an end date.

Parámetros:

end_date (date | datetime | str | None) – Optional response date.

Devuelve:

Datetime assigned to submitted fact rows.

Tipo del valor devuelto:

datetime

database.repositories._date_label(end_date)

Return a stable ISO label for an end date.

Parámetros:

end_date (date | datetime | str | None) – Optional cutoff date.

Devuelve:

ISO date or datetime label.

Tipo del valor devuelto:

str

database.repositories._end_date_filter(statement, end_date)

Apply an inclusive end-date filter to a fact-response statement.

Parámetros:

  • statement (sqlalchemy.Select) – SQLAlchemy select statement.

  • end_date (date | datetime | str | None) – Optional cutoff date.

Devuelve:

Statement filtered by date_time.

Tipo del valor devuelto:

sqlalchemy.Select

database.repositories._municipality_or_raise(session, code)

Return a municipality by code or raise a clear error.

Parámetros:

  • session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

  • code (str) – Municipal code.

Devuelve:

Matching municipality ORM row.

Muestra:

ValueError – If the municipality code is unknown.

Tipo del valor devuelto:

DMMunicipality

database.repositories._municipality_to_dict(municipality)

Convert a municipality ORM row into the app dictionary shape.

Parámetros:

municipality (DMMunicipality) – Municipality ORM row.

Devuelve:

Municipality dictionary.

Tipo del valor devuelto:

dict[str, Any]

database.repositories._service_to_dict(service)

Convert a service ORM row into the app dictionary shape.

Parámetros:

service (DMService) – Service ORM row.

Devuelve:

Service dictionary.

Tipo del valor devuelto:

dict[str, Any]

database.repositories._applicable_service_ids(municipality)

Return service ids that apply to a municipality.

Parámetros:

municipality (DMMunicipality) – Municipality ORM row.

Devuelve:

Set of applicable service ids.

Tipo del valor devuelto:

set[int]

database.repositories._latest_response_values(session, municipality_id, end_date=None)

Return latest numeric values by indicator code for a municipality.

Parámetros:

  • session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

  • municipality_id (int) – Municipality primary key.

  • end_date (date | datetime | str | None) – Optional inclusive cutoff date.

Devuelve:

Mapping from indicator code to numeric value.

Tipo del valor devuelto:

dict[str, float]

database.repositories._count_indicators(session)

Count configured IGSM indicators.

Parámetros:

session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

Devuelve:

Number of indicator rows.

Tipo del valor devuelto:

int

database.repositories._count_municipalities(session)

Count configured municipalities.

Parámetros:

session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

Devuelve:

Number of municipality rows.

Tipo del valor devuelto:

int

database.repositories._count_distinct_response_pairs(session, end_date=None, municipality_id=None)

Count unique municipality-indicator response pairs.

Multiple fact rows for the same municipality and indicator before the same cutoff date count as one submitted response.

Parámetros:

  • session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

  • end_date (date | datetime | str | None) – Optional inclusive cutoff date.

  • municipality_id (int | None) – Optional municipality primary key filter.

Devuelve:

Number of distinct response pairs.

Tipo del valor devuelto:

int

database.repositories._completion_percentage(responses_received, responses_expected)

Calculate a data-completion percentage.

Parámetros:

  • responses_received (int) – Count of unique submitted responses.

  • responses_expected (int) – Count of expected responses.

Devuelve:

Completion percentage rounded to two decimals.

Tipo del valor devuelto:

float

database.repositories.list_municipalities(session=None)

List municipalities in display order.

Parámetros:

session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Municipality dictionaries.

Tipo del valor devuelto:

list[dict[str, Any]]

database.repositories.get_municipality_by_code(code, session=None)

Return one municipality by code.

Parámetros:

  • code (str) – Municipal code.

  • session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Municipality dictionary, or None.

Tipo del valor devuelto:

dict[str, Any] | None

database.repositories.get_municipality_by_name(name, session=None)

Return one municipality by exact name.

Parámetros:

  • name (str) – Municipality name.

  • session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Municipality dictionary, or None.

Tipo del valor devuelto:

dict[str, Any] | None

database.repositories.get_municipality_names(session=None)

Return municipality names.

Parámetros:

session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Sorted municipality names.

Tipo del valor devuelto:

list[str]

database.repositories.list_municipalities_by_region(region, session=None)

List municipalities for a region.

Parámetros:

  • region (str) – Region name.

  • session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Municipality dictionaries.

Tipo del valor devuelto:

list[dict[str, Any]]

database.repositories.get_services_for_municipality(municipality_code, session=None)

Return services applicable to one municipality.

Parámetros:

  • municipality_code (str) – Municipal code.

  • session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Mapping from service name to service dictionary.

Tipo del valor devuelto:

dict[str, dict[str, Any]]

database.repositories.get_indicators_for_service(service_code, session=None)

Return indicators for a service code.

Parámetros:

  • service_code (str) – Service code.

  • session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Indicator dictionaries.

Tipo del valor devuelto:

list[dict[str, Any]]

database.repositories.get_latest_stage_weights(end_date=None, session=None)

Return the latest stage weights as of a date.

Parámetros:

  • end_date (EndDate) – Inclusive cutoff date. Defaults to today.

  • session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Stage weights keyed by Spanish stage name.

Tipo del valor devuelto:

dict[str, float]

database.repositories.get_current_stage_weights(as_of=None, session=None)

Return the effective stage weights.

Parámetros:

  • as_of (date | None) – Effective date. Defaults to today.

  • session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Stage weights keyed by Spanish stage name.

Tipo del valor devuelto:

dict[str, float]

database.repositories.save_stage_weights(planning, execution, evaluation, effective_from, description=None, session=None)

Save a new effective-dated stage-weight row.

Parámetros:

  • planning (float) – Planning weight.

  • execution (float) – Execution weight.

  • evaluation (float) – Evaluation weight.

  • effective_from (date) – Effective date.

  • description (str | None) – Ignored compatibility field; no app metadata table exists.

  • session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Saved stage-weight fields.

Muestra:

ValueError – If weights do not sum to 1.0.

Tipo del valor devuelto:

dict[str, Any]

database.repositories.get_latest_maturity_thresholds(end_date=None, session=None)

Return the latest maturity thresholds as of a date.

Parámetros:

  • end_date (EndDate) – Inclusive cutoff date. Defaults to today.

  • session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Threshold values keyed by maturity boundary name.

Tipo del valor devuelto:

dict[str, float]

database.repositories.get_current_maturity_thresholds(as_of=None, session=None)

Return the effective maturity thresholds.

Parámetros:

  • as_of (date | None) – Effective date. Defaults to today.

  • session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Threshold values keyed by maturity boundary name.

Tipo del valor devuelto:

dict[str, float]

database.repositories.submit_indicator_responses(municipality_code, end_date, responses, evidence=None, actor_subject=None, session=None)

Persist numeric indicator responses as fact rows.

Parámetros:

  • municipality_code (str) – Municipal code.

  • end_date (EndDate) – Response date assigned to the fact rows.

  • responses (dict[str, Any]) – Mapping from indicator code to submitted value.

  • evidence (dict[str, Any] | None) – Ignored compatibility field; evidence metadata is external.

  • actor_subject (str | None) – Ignored compatibility field; identity is external.

  • session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Submission-style summary containing fact ids and row counts.

Tipo del valor devuelto:

dict[str, Any]

database.repositories.get_latest_responses_for_municipality(municipality_code, end_date=None, session=None)

Return latest fact responses for a municipality.

Parámetros:

  • municipality_code (str) – Municipal code.

  • end_date (EndDate) – Optional inclusive cutoff date.

  • session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Mapping from indicator code to value.

Tipo del valor devuelto:

dict[str, Any]

database.repositories.get_latest_indicator_responses(end_date=None, session=None)

Return latest fact responses for all municipalities.

Parámetros:

  • end_date (EndDate) – Optional inclusive cutoff date.

  • session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Latest response rows by municipality and indicator.

Tipo del valor devuelto:

list[dict[str, Any]]

database.repositories.get_national_statistics(end_date=None, session=None)

Return national data-completion statistics.

Parámetros:

  • end_date (EndDate) – Optional inclusive cutoff date.

  • session (Session | None) – Optional SQLAlchemy session.

Devuelve:

National completion statistics dictionary.

Tipo del valor devuelto:

dict[str, Any]

database.repositories.get_municipality_completion_statistics(municipality_code, end_date=None, session=None)

Return data-completion statistics for one municipality.

Parámetros:

  • municipality_code (str) – Municipal code.

  • end_date (EndDate) – Optional inclusive cutoff date.

  • session (Session | None) – Optional SQLAlchemy session.

Devuelve:

Municipality completion statistics dictionary.

Tipo del valor devuelto:

dict[str, Any]

Seed helpers for the IGSM ORM package.

database.seed.seed_all(include_demo_results=False, session=None)

Seed all reference data into the configured database.

Parámetros:

  • include_demo_results (bool) – Deprecated compatibility flag.

  • session (Session | None) – Optional caller-managed SQLAlchemy session.

Tipo del valor devuelto:

None

database.seed.seed_reference_data(session)

Seed the complete IGSM reference dataset.

Parámetros:

session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

Tipo del valor devuelto:

None

database.seed.seed_municipalities(session)

Seed municipality rows and diversified-service links.

Parámetros:

session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

Tipo del valor devuelto:

None

database.seed.seed_igsm_structure(session)

Seed IGSM axes, services, stages, and indicators from source CSV.

Parámetros:

session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

Tipo del valor devuelto:

None

database.seed.seed_stage_weights(session)

Seed the default stage weights.

Parámetros:

session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

Tipo del valor devuelto:

None

database.seed.seed_maturity_thresholds(session)

Seed the default maturity thresholds.

Parámetros:

session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

Tipo del valor devuelto:

None

database.seed.seed_demo_results(session, period='2025')

Keep the legacy demo hook as a no-op after app tables were removed.

Parámetros:

  • session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

  • period (str) – Requested demo period.

Tipo del valor devuelto:

None

SQLAlchemy engine and session helpers.

database.session._engine_kwargs(database_url)

Build SQLAlchemy engine keyword arguments.

Parámetros:

database_url (str) – SQLAlchemy database URL.

Devuelve:

Keyword arguments for create_engine.

Tipo del valor devuelto:

dict

database.session.get_engine(database_url=None)

Return a cached SQLAlchemy engine.

Parámetros:

database_url (str | None) – Optional database URL override.

Devuelve:

SQLAlchemy engine bound to the requested database URL.

Tipo del valor devuelto:

sqlalchemy.engine.Engine

database.session.get_session_factory(database_url=None)

Return a cached SQLAlchemy session factory.

Parámetros:

database_url (str | None) – Optional database URL override.

Devuelve:

Session factory bound to the cached engine.

Tipo del valor devuelto:

sqlalchemy.orm.sessionmaker.sqlalchemy.orm.Session

database.session.session_scope(database_url=None)

Provide a transactional session scope.

Parámetros:

database_url (str | None) – Optional database URL override.

Campos:

Active SQLAlchemy session.

Muestra:

Exception – Re-raises any exception after rolling back the transaction.

Tipo del valor devuelto:

Iterator[sqlalchemy.orm.Session]

Import the CSV source files as the ORM baseline dataset.

class database.import_source_baseline.SourceIndicator(code, name, grouping, axis, axis_code, service, service_code, stage, diversified_key)

Bases: object

Indicator row parsed from the source dictionary CSV.

Variables:

  • code (str) – Official indicator code.

  • name (str) – Indicator display name.

  • grouping (str) – Normalized grouping label.

  • axis (str) – Axis display name.

  • axis_code (str) – Official axis code.

  • service (str) – Service display name.

  • service_code (str) – Official service code derived from the indicator code.

  • stage (str) – Stage display name.

  • diversified_key (str | None) – Diversified-service key when applicable.

Parámetros:

  • code (str)

  • name (str)

  • grouping (str)

  • axis (str)

  • axis_code (str)

  • service (str)

  • service_code (str)

  • stage (str)

  • diversified_key (str | None)

code: str
name: str
grouping: str
axis: str
axis_code: str
service: str
service_code: str
stage: str
diversified_key: str | None
class database.import_source_baseline.SourceValue(municipality_code, indicator_code, value)

Bases: object

Numeric source response parsed from the results CSV.

Variables:

  • municipality_code (str) – Municipality code matched from source labels.

  • indicator_code (str) – Indicator code.

  • value (float) – Numeric response value.

Parámetros:

  • municipality_code (str)

  • indicator_code (str)

  • value (float)

municipality_code: str
indicator_code: str
value: float
class database.import_source_baseline.SourceMunicipality(code, name, province, region, latitude, longitude, diversified_keys)

Bases: object

Municipality metadata parsed from baseline results.

Variables:

  • code (str) – Municipality code.

  • name (str) – Municipality name from the source or reference catalog.

  • province (str | None) – Province name.

  • region (str | None) – Planning region name.

  • latitude (float | None) – Latitude from the reference catalog.

  • longitude (float | None) – Longitude from the reference catalog.

  • diversified_keys (tuple[str, ...]) – Diversified-service keys assigned to the municipality.

Parámetros:

  • code (str)

  • name (str)

  • province (str | None)

  • region (str | None)

  • latitude (float | None)

  • longitude (float | None)

  • diversified_keys (tuple[str, ...])

code: str
name: str
province: str | None
region: str | None
latitude: float | None
longitude: float | None
diversified_keys: tuple[str, ...]
database.import_source_baseline.load_source_baseline(period='2025', replace=True, source_dir=None, database_url=None, dry_run=False)

Load the source CSV files into the configured ORM database.

The import is intentionally idempotent by default: previous fact rows created by this source-baseline importer are replaced while unrelated facts are left untouched.

Parámetros:

  • period (str) – Submission period label assigned to the summary.

  • replace (bool) – Whether to replace existing baseline fact rows.

  • source_dir (str | Path | None) – Optional directory containing the source CSV files.

  • database_url (str | None) – Optional database URL override.

  • dry_run (bool) – Whether to validate and summarize without writing to the database.

Devuelve:

Import summary with source counts, database counts, and write counts.

Muestra:

  • ValueError – If the CSV contents are invalid or reference unknown codes.

  • FileNotFoundError – If required source CSV files are missing.

Tipo del valor devuelto:

dict[str, Any]

database.import_source_baseline._read_dictionary(path)

Read and validate the indicator dictionary CSV.

Parámetros:

path (Path) – Path to dictionary.csv.

Devuelve:

Parsed source indicator rows.

Muestra:

  • ValueError – If required fields are missing or invalid.

  • FileNotFoundError – If the CSV file is missing.

Tipo del valor devuelto:

list[SourceIndicator]

database.import_source_baseline._read_results(path, indicators_by_code, municipality_lookup)

Read and validate baseline result values.

Parámetros:

  • path (Path) – Path to the long-form results CSV.

  • indicators_by_code (dict[str, SourceIndicator]) – Source indicators keyed by indicator code.

  • municipality_lookup (dict[str, str]) – Municipality code lookup keyed by normalized labels.

Devuelve:

Parsed source values and row-count statistics.

Muestra:

  • ValueError – If the file references unknown indicators or nonnumeric values.

  • FileNotFoundError – If the CSV file is missing.

Tipo del valor devuelto:

tuple[list[SourceValue], dict[str, int]]

database.import_source_baseline._read_csv_rows(path, required_columns)

Read a CSV file as dictionaries and validate required columns.

Parámetros:

  • path (Path) – CSV file path.

  • required_columns (set[str]) – Column names that must exist in the file.

Devuelve:

CSV rows with missing cell values normalized to empty strings.

Muestra:

  • FileNotFoundError – If the file does not exist.

  • ValueError – If required columns are missing.

Tipo del valor devuelto:

list[dict[str, str]]

database.import_source_baseline._service_code_from_indicator_code(code)

Derive a service code from an indicator code.

Parámetros:

code (str) – Dot-separated indicator code.

Devuelve:

First three code segments joined as a service code.

Muestra:

ValueError – If the indicator code has fewer than three segments.

Tipo del valor devuelto:

str

database.import_source_baseline._normalize_name(value)

Normalize a municipality label for matching.

Parámetros:

value (str) – Raw municipality label.

Devuelve:

Lowercase ASCII label with punctuation and extra whitespace removed.

Tipo del valor devuelto:

str

database.import_source_baseline._reference_municipalities_by_normalized_name()

Build the reference municipality lookup by normalized name.

Devuelve:

Municipality dictionaries keyed by normalized aliases.

Tipo del valor devuelto:

dict[str, dict[str, Any]]

database.import_source_baseline._source_municipalities_from_results(path)

Extract municipality metadata and lookup aliases from results.

Parámetros:

path (Path) – Path to the long-form results CSV.

Devuelve:

Source municipalities keyed by code and normalized-name lookup mapping.

Muestra:

  • ValueError – If a source municipality cannot be matched.

  • FileNotFoundError – If the CSV file is missing.

Tipo del valor devuelto:

tuple[dict[str, SourceMunicipality], dict[str, str]]

database.import_source_baseline._municipality_code_for_result(row, municipality_lookup)

Resolve a municipality code for one result row.

Parámetros:

  • row (dict[str, str]) – CSV result row.

  • municipality_lookup (dict[str, str]) – Municipality code lookup keyed by normalized labels.

Devuelve:

Matched municipality code.

Muestra:

ValueError – If the row cannot be matched to a municipality.

Tipo del valor devuelto:

str

database.import_source_baseline._baseline_fact_count(session)

Count fact rows created by the baseline importer.

Parámetros:

session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

Devuelve:

Number of fact rows using the baseline import timestamp.

Tipo del valor devuelto:

int

database.import_source_baseline._delete_existing_baseline(session)

Delete fact rows created by a previous baseline import.

Parámetros:

session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

Devuelve:

Summary counts for deleted rows.

Tipo del valor devuelto:

dict[str, int]

database.import_source_baseline._derive_diversified_keys(values, dictionary_by_code)

Infer diversified-service keys from submitted source values.

Parámetros:

  • values (list[SourceValue]) – Parsed source response values.

  • dictionary_by_code (dict[str, SourceIndicator]) – Source indicators keyed by indicator code.

Devuelve:

Diversified-service keys keyed by municipality code.

Tipo del valor devuelto:

dict[str, set[str]]

database.import_source_baseline._upsert_municipalities(session, source_municipalities, derived_diversified)

Insert or update municipalities and diversified-service links.

Parámetros:

  • session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

  • source_municipalities (dict[str, SourceMunicipality]) – Municipality metadata keyed by municipal code.

  • derived_diversified (dict[str, set[str]]) – Diversified keys inferred from baseline values.

Devuelve:

Set of municipal codes touched by the import.

Tipo del valor devuelto:

set[str]

database.import_source_baseline._upsert_igsm_structure(session, dictionary)

Insert or update axes, services, stages, and indicators.

Parámetros:

  • session (sqlalchemy.orm.Session) – Active SQLAlchemy session.

  • dictionary (list[SourceIndicator]) – Parsed source dictionary rows.

Devuelve:

Current table counts for the IGSM structure.

Tipo del valor devuelto:

dict[str, int]

database.import_source_baseline.main()

Run the source-baseline importer CLI.

Tipo del valor devuelto:

None