ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix/un-alembic

Browse Source
This commit is contained in:
hreyes
2026-03-24 13:15:15 -06:00
parent b5b99a5ddb
commit 3d1c73565f
2 changed files with 58 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
backend/alembic/versions/c1a2b3d4e5f6_create_core_task_runs.py
View File

@@ -1,4 +1,4 @@
"""create_core_task_runs
"""create_core_task_runs and backfill code_pedimento_regimens
Revision ID: c1a2b3d4e5f6
Revises: bccb7f8986c7
@@ -11,6 +11,13 @@ from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from api.v1.modules.public.reference_data.code_pedimento_regimens.seed import (
seed as code_pedimento_regimens_seed_full,
)
from api.v1.modules.public.reference_data.pedimento_codes.seed import (
seed as pedimento_codes_seed,
)
# revision identifiers, used by Alembic.
revision: str = "c1a2b3d4e5f6"
@@ -19,6 +26,12 @@ branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def _format_value(val: str | None) -> str:
if val is None or str(val).strip() == "" or str(val).upper() == "NONE":
return "NULL"
return f"'{str(val).replace(chr(39), chr(39) * 2)}'"
def upgrade() -> None:
op.create_table(
"task_runs",
@@ -75,6 +88,50 @@ def upgrade() -> None:
schema="core",
)
pedimento_desc_by_code = {code: desc for code, desc in pedimento_codes_seed}
required_pedimento_codes = sorted(
{pedimento_code for pedimento_code, _regimen_code, _type_code in code_pedimento_regimens_seed_full}
)
values_missing_codes = ", ".join(
[
f"({_format_value(code)}, {_format_value(pedimento_desc_by_code.get(code, f'AUTO-GENERATED FOR FK ({code})'))})"
for code in required_pedimento_codes
]
)
op.execute(
f"""
INSERT INTO public.pedimento_codes (code, description)
VALUES {values_missing_codes}
ON CONFLICT (code) DO NOTHING;
"""
)
values_relations = ", ".join(
[
f"({_format_value(pedimento_code)}, {_format_value(regimen_code)}, {_format_value(type_code)})"
for pedimento_code, regimen_code, type_code in code_pedimento_regimens_seed_full
]
)
op.execute(
f"""
INSERT INTO public.code_pedimento_regimens (pedimento_code, regimen_code, type_code)
SELECT src.pedimento_code, src.regimen_code, src.type_code
FROM (VALUES {values_relations}) AS src(pedimento_code, regimen_code, type_code)
JOIN public.pedimento_codes pc
ON pc.code = src.pedimento_code
JOIN public.pedimento_regimens pr
ON pr.code = src.regimen_code
WHERE NOT EXISTS (
SELECT 1
FROM public.code_pedimento_regimens cpr
WHERE cpr.pedimento_code = src.pedimento_code
AND cpr.regimen_code = src.regimen_code
AND COALESCE(cpr.type_code, '') = COALESCE(src.type_code, '')
);
"""
)
def downgrade() -> None:
op.drop_index("ix_core_task_runs_tenant_company_updated", table_name="task_runs", schema="core")

82
backend/alembic/versions/d4e5f6a7b8c9_backfill_full_code_pedimento_regimens.py
View File

@@ -1,82 +0,0 @@
"""backfill_full_code_pedimento_regimens
Revision ID: d4e5f6a7b8c9
Revises: c1a2b3d4e5f6
Create Date: 2026-03-24 13:10:00.000000
"""
from typing import Sequence, Union
from alembic import op
from api.v1.modules.public.reference_data.code_pedimento_regimens.seed import (
seed as code_pedimento_regimens_seed_full,
)
from api.v1.modules.public.reference_data.pedimento_codes.seed import (
seed as pedimento_codes_seed,
)
# revision identifiers, used by Alembic.
revision: str = "d4e5f6a7b8c9"
down_revision: Union[str, Sequence[str], None] = "c1a2b3d4e5f6"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def _format_value(val: str | None) -> str:
if val is None or str(val).strip() == "" or str(val).upper() == "NONE":
return "NULL"
return f"'{str(val).replace(chr(39), chr(39) * 2)}'"
def upgrade() -> None:
"""Backfill full code_pedimento_regimens data without editing old migrations."""
pedimento_desc_by_code = {code: desc for code, desc in pedimento_codes_seed}
required_pedimento_codes = sorted(
{pedimento_code for pedimento_code, _regimen_code, _type_code in code_pedimento_regimens_seed_full}
)
values_missing_codes = ", ".join(
[
f"({_format_value(code)}, {_format_value(pedimento_desc_by_code.get(code, f'AUTO-GENERATED FOR FK ({code})'))})"
for code in required_pedimento_codes
]
)
op.execute(
f"""
INSERT INTO public.pedimento_codes (code, description)
VALUES {values_missing_codes}
ON CONFLICT (code) DO NOTHING;
"""
)
values_relations = ", ".join(
[
f"({_format_value(pedimento_code)}, {_format_value(regimen_code)}, {_format_value(type_code)})"
for pedimento_code, regimen_code, type_code in code_pedimento_regimens_seed_full
]
)
op.execute(
f"""
INSERT INTO public.code_pedimento_regimens (pedimento_code, regimen_code, type_code)
SELECT src.pedimento_code, src.regimen_code, src.type_code
FROM (VALUES {values_relations}) AS src(pedimento_code, regimen_code, type_code)
JOIN public.pedimento_codes pc
ON pc.code = src.pedimento_code
JOIN public.pedimento_regimens pr
ON pr.code = src.regimen_code
WHERE NOT EXISTS (
SELECT 1
FROM public.code_pedimento_regimens cpr
WHERE cpr.pedimento_code = src.pedimento_code
AND cpr.regimen_code = src.regimen_code
AND COALESCE(cpr.type_code, '') = COALESCE(src.type_code, '')
);
"""
)
def downgrade() -> None:
"""Preserve seeded data on downgrade."""