"""HR multi-country public-holiday calendar + 2026 baseline seed. Helix had no per-country holiday model at all: ``app/core/working_hours.py`` and ``app/domains/attendance/facts.py`` both carry a comment saying so, and the only holiday table in the codebase (``finance_close_holidays``) is the month-end close engine's China-only workday calendar, which stays untouched here. Two tables: * ``hr_public_holidays`` — one row per (country, day, holiday name). The natural key is deliberately three-part, not (country, day): Hong Kong observes both Ching Ming Festival and Holy Saturday on 2026-04-04, and Romania observes both Children's Day and Orthodox Pentecost Monday on 2026-06-01. A (country, day) unique key would silently drop one of each pair. * ``hr_public_holiday_calendars`` — per (country, year) metadata (HR's headline day count, Google Calendar link, regional-holiday advisory). A missing row is the "not maintained" signal the External API reports as ``no_data``. Seeded with the 22 countries/regions of the HR-owned 2026 document (322 holiday dates + China's 6 makeup workdays) from :mod:`app.domains.hr.holiday_seed_2026`. The seed is idempotent: rows are keyed on (country, day, name_key) and skipped when already present, so re-running never duplicates and never overwrites an edit HR has since made. No FKs: ``created_by_id`` / ``updated_by_id`` are plain id columns sized via ``id_type_for('identity_persons')`` per the platform's TiDB-FK-safe convention. ``name_key_for`` is imported from :mod:`app.domains.hr.holiday_keys`, not from the models module: ``holiday_models`` imports ``app.core.database``, which builds ``Settings`` from env vars that ``scripts/check_alembic_chain.py`` does not set — importing it here crashes the CI migration guard on file load alone. Revision ID: 462 Revises: 461 """ from __future__ import annotations import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for, table_kw_for from app.domains.hr.holiday_keys import name_key_for from app.domains.hr.holiday_seed_2026 import SEED_2026, SEED_YEAR revision = "462" down_revision = "461" branch_labels = None depends_on = None HOLIDAYS = "hr_public_holidays" CALENDARS = "hr_public_holiday_calendars" def _table_exists(bind, name: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = :n" ), {"n": name}, ).scalar() is not None def upgrade() -> None: bind = op.get_bind() actor_id_type = id_type_for("identity_persons") tkw = table_kw_for("identity_persons") if not _table_exists(bind, HOLIDAYS): op.create_table( HOLIDAYS, sa.Column("id", sa.BigInteger, nullable=False, autoincrement=True), sa.Column("country", sa.String(8), nullable=False), sa.Column("day", sa.Date, nullable=False), sa.Column("group_key", sa.String(96), nullable=False), sa.Column("name_en", sa.String(255), nullable=False), sa.Column("name_local", sa.String(255), nullable=True), sa.Column("name_key", sa.String(128), nullable=False), sa.Column("day_type", sa.String(24), nullable=False, server_default="public_holiday"), sa.Column("note", sa.String(500), nullable=True), sa.Column("source", sa.String(16), nullable=False, server_default="manual"), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("created_by_id", actor_id_type, nullable=True), sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("updated_by_id", actor_id_type, nullable=True), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("country", "day", "name_key", name="uq_hr_public_holiday_country_day_name"), **tkw, ) op.create_index("ix_hr_public_holiday_country_day", HOLIDAYS, ["country", "day"]) op.create_index("ix_hr_public_holiday_country_group", HOLIDAYS, ["country", "group_key"]) op.create_index("ix_hr_public_holiday_day", HOLIDAYS, ["day"]) if not _table_exists(bind, CALENDARS): op.create_table( CALENDARS, sa.Column("id", sa.BigInteger, nullable=False, autoincrement=True), sa.Column("country", sa.String(8), nullable=False), sa.Column("year", sa.Integer, nullable=False), sa.Column("label", sa.String(64), nullable=True), sa.Column("stated_days", sa.Integer, nullable=True), sa.Column("gcal_url", sa.String(512), nullable=True), sa.Column("region_note", sa.Text, nullable=True), sa.Column("source", sa.String(16), nullable=False, server_default="manual"), sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("created_by_id", actor_id_type, nullable=True), sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), sa.Column("updated_by_id", actor_id_type, nullable=True), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("country", "year", name="uq_hr_public_holiday_cal_country_year"), **tkw, ) op.create_index("ix_hr_public_holiday_cal_year", CALENDARS, ["year"]) _seed(bind) def _seed(bind) -> None: """Insert the 2026 baseline, skipping anything already present. Committed in per-country batches rather than one big statement: TiDB bills a write transaction's whole membuffer to the *current* statement, so a single 328-row insert plus its existence probes is a needlessly fat transaction for a migration that may run against production. """ for country, spec in SEED_2026.items(): existing_cal = bind.execute( sa.text( f"SELECT 1 FROM {CALENDARS} WHERE country = :c AND year = :y" ), {"c": country, "y": SEED_YEAR}, ).scalar() if not existing_cal: bind.execute( sa.text( f"INSERT INTO {CALENDARS} " "(country, year, label, stated_days, gcal_url, region_note, source) " "VALUES (:c, :y, :label, :stated, :gcal, :note, 'seed')" ), { "c": country, "y": SEED_YEAR, "label": spec["label"], "stated": spec["stated_days"], "gcal": spec["gcal_url"], "note": spec["region_note"], }, ) present = { (row[0].isoformat(), row[1]) for row in bind.execute( sa.text( f"SELECT day, name_key FROM {HOLIDAYS} WHERE country = :c" ), {"c": country}, ).fetchall() } rows = [] for day, group_key, name_en, name_local, day_type in spec["days"]: key = name_key_for(name_en) if (day, key) in present: continue present.add((day, key)) rows.append({ "c": country, "day": day, "gk": group_key, "en": name_en, "local": name_local, "nk": key, "dt": day_type, }) if rows: bind.execute( sa.text( f"INSERT INTO {HOLIDAYS} " "(country, day, group_key, name_en, name_local, name_key, " " day_type, source) " "VALUES (:c, :day, :gk, :en, :local, :nk, :dt, 'seed')" ), rows, ) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, CALENDARS): op.drop_table(CALENDARS) if _table_exists(bind, HOLIDAYS): op.drop_table(HOLIDAYS)