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

feat(states): add initial seed data and update State model with foreign key

Browse Source
This commit is contained in:
acazares
2026-01-20 13:42:00 -06:00
parent 673467ec70
commit 9a0cc4487d
3 changed files with 216 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

168
backend/alembic/versions/7937209f9718_seed_initial_data.py
View File

@@ -44,6 +44,7 @@ from api.v1.modules.public.reference_data.pedimento_regimens.seed import (
seed as pedimento_regimens_seed,
)
from api.v1.modules.public.reference_data.sectors.seed import seed as sectors_seed
from api.v1.modules.public.reference_data.states.seed import seed as states_seed
from api.v1.modules.public.reference_data.transport_modes.seed import (
seed as transport_modes_seed,
)
@@ -53,11 +54,21 @@ from api.v1.modules.public.reference_data.transport_types.seed import (
from api.v1.modules.public.reference_data.valuation_methods.seed import (
seed as valuation_methods_seed,
)
from api.v1.modules.a76.general_catalogs.units_of_measure.seed import seed as units_of_measure_seed
from api.v1.modules.a76.general_catalogs.units_of_measure.seed_ace import seed as ace_seed
from api.v1.modules.a76.general_catalogs.units_of_measure.seed_oma import seed as oma_seed
from api.v1.modules.a76.general_catalogs.units_of_measure.seed_ame import seed as ame_seed
from api.v1.modules.a76.general_catalogs.units_of_measure.seed_adua import seed as adua_seed
from api.v1.modules.a76.general_catalogs.units_of_measure.seed import (
seed as units_of_measure_seed,
)
from api.v1.modules.a76.general_catalogs.units_of_measure.seed_ace import (
seed as ace_seed,
)
from api.v1.modules.a76.general_catalogs.units_of_measure.seed_oma import (
seed as oma_seed,
)
from api.v1.modules.a76.general_catalogs.units_of_measure.seed_ame import (
seed as ame_seed,
)
from api.v1.modules.a76.general_catalogs.units_of_measure.seed_adua import (
seed as adua_seed,
)
from api.v1.modules.core.permissions.seed import (
seed_invoices,
seed_user,
@@ -73,12 +84,12 @@ depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
"""Upgrade schema."""
# --- UTILIDAD DE FORMATEO ---
def format_value(val):
if val is None or str(val).strip() == '' or str(val).upper() == 'NONE':
return 'NULL'
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)}'"
# --- SEEDS PUBLIC (Tablas base) ---
@@ -310,93 +321,146 @@ def upgrade() -> None:
# --- SEEDS A76 (Unidades de Medida) ---
# ACE
val_ace = ", ".join([f"({format_value(c)}, {format_value(d)})" for c, d in ace_seed])
op.execute(f"INSERT INTO a76.unit_of_measure_ace (code, description) VALUES {val_ace} ON CONFLICT ON CONSTRAINT uq_uom_ace_code DO NOTHING;")
val_ace = ", ".join(
[f"({format_value(c)}, {format_value(d)})" for c, d in ace_seed]
)
op.execute(
f"INSERT INTO a76.unit_of_measure_ace (code, description) VALUES {val_ace} ON CONFLICT ON CONSTRAINT uq_uom_ace_code DO NOTHING;"
)
# OMA
val_oma = ", ".join([f"({format_value(c)}, {format_value(d)})" for c, d in oma_seed])
op.execute(f"INSERT INTO a76.unit_of_measure_oma (code, description) VALUES {val_oma} ON CONFLICT ON CONSTRAINT uq_uom_oma_code DO NOTHING;")
val_oma = ", ".join(
[f"({format_value(c)}, {format_value(d)})" for c, d in oma_seed]
)
op.execute(
f"INSERT INTO a76.unit_of_measure_oma (code, description) VALUES {val_oma} ON CONFLICT ON CONSTRAINT uq_uom_oma_code DO NOTHING;"
)
# AME
val_ame = ", ".join([f"({format_value(c)}, {format_value(d)})" for c, d in ame_seed])
op.execute(f"INSERT INTO a76.unit_of_measure_american (code, description) VALUES {val_ame} ON CONFLICT ON CONSTRAINT uq_uom_american_code DO NOTHING;")
val_ame = ", ".join(
[f"({format_value(c)}, {format_value(d)})" for c, d in ame_seed]
)
op.execute(
f"INSERT INTO a76.unit_of_measure_american (code, description) VALUES {val_ame} ON CONFLICT ON CONSTRAINT uq_uom_american_code DO NOTHING;"
)
# ADUA (Customs)
val_adua = ", ".join([f"({format_value(c)}, {format_value(d)})" for c, d in adua_seed])
op.execute(f"INSERT INTO a76.unit_of_measure_customs (code, description) VALUES {val_adua} ON CONFLICT ON CONSTRAINT uq_uom_customs_code DO NOTHING;")
val_adua = ", ".join(
[f"({format_value(c)}, {format_value(d)})" for c, d in adua_seed]
)
op.execute(
f"INSERT INTO a76.unit_of_measure_customs (code, description) VALUES {val_adua} ON CONFLICT ON CONSTRAINT uq_uom_customs_code DO NOTHING;"
)
# Recolectar códigos adicionales que faltan en los catálogos
additional_customs = set()
additional_american = set()
additional_ace = set()
additional_oma = set()
existing_customs = {c for c, d in adua_seed}
existing_american = {c for c, d in ame_seed}
existing_ace = {c for c, d in ace_seed}
existing_oma = {c for c, d in oma_seed}
for code, desc, desc_en, customs, american, ace, oma in units_of_measure_seed:
if customs and customs.strip() and customs not in existing_customs:
additional_customs.add((customs, f'Auto-generated from {code}'))
additional_customs.add((customs, f"Auto-generated from {code}"))
if american and american.strip() and american not in existing_american:
additional_american.add((american, f'Auto-generated from {code}'))
additional_american.add((american, f"Auto-generated from {code}"))
if ace and ace.strip() and ace not in existing_ace:
additional_ace.add((ace, f'Auto-generated from {code}'))
additional_ace.add((ace, f"Auto-generated from {code}"))
if oma and oma.strip() and oma not in existing_oma:
additional_oma.add((oma, f'Auto-generated from {code}'))
additional_oma.add((oma, f"Auto-generated from {code}"))
# Insertar códigos adicionales
if additional_customs:
val_add_customs = ", ".join([f"({format_value(c)}, {format_value(d)})" for c, d in additional_customs])
op.execute(f"INSERT INTO a76.unit_of_measure_customs (code, description, a76_unit_code) VALUES {val_add_customs} ON CONFLICT ON CONSTRAINT uq_uom_customs_code DO NOTHING;")
val_add_customs = ", ".join(
[f"({format_value(c)}, {format_value(d)})" for c, d in additional_customs]
)
op.execute(
f"INSERT INTO a76.unit_of_measure_customs (code, description, a76_unit_code) VALUES {val_add_customs} ON CONFLICT ON CONSTRAINT uq_uom_customs_code DO NOTHING;"
)
if additional_american:
val_add_american = ", ".join([f"({format_value(c)}, {format_value(d)})" for c, d in additional_american])
op.execute(f"INSERT INTO a76.unit_of_measure_american (code, description) VALUES {val_add_american} ON CONFLICT ON CONSTRAINT uq_uom_american_code DO NOTHING;")
val_add_american = ", ".join(
[f"({format_value(c)}, {format_value(d)})" for c, d in additional_american]
)
op.execute(
f"INSERT INTO a76.unit_of_measure_american (code, description) VALUES {val_add_american} ON CONFLICT ON CONSTRAINT uq_uom_american_code DO NOTHING;"
)
if additional_ace:
val_add_ace = ", ".join([f"({format_value(c)}, {format_value(d)})" for c, d in additional_ace])
op.execute(f"INSERT INTO a76.unit_of_measure_ace (code, description) VALUES {val_add_ace} ON CONFLICT ON CONSTRAINT uq_uom_ace_code DO NOTHING;")
val_add_ace = ", ".join(
[f"({format_value(c)}, {format_value(d)})" for c, d in additional_ace]
)
op.execute(
f"INSERT INTO a76.unit_of_measure_ace (code, description) VALUES {val_add_ace} ON CONFLICT ON CONSTRAINT uq_uom_ace_code DO NOTHING;"
)
if additional_oma:
val_add_oma = ", ".join([f"({format_value(c)}, {format_value(d)})" for c, d in additional_oma])
op.execute(f"INSERT INTO a76.unit_of_measure_oma (code, description) VALUES {val_add_oma} ON CONFLICT ON CONSTRAINT uq_uom_oma_code DO NOTHING;")
val_add_oma = ", ".join(
[f"({format_value(c)}, {format_value(d)})" for c, d in additional_oma]
)
op.execute(
f"INSERT INTO a76.unit_of_measure_oma (code, description) VALUES {val_add_oma} ON CONFLICT ON CONSTRAINT uq_uom_oma_code DO NOTHING;"
)
# TABLA MAESTRA UOM
# TODO: Generar tenant_id y company_id correctos
val_uom = ", ".join([
f"({format_value(code)}, {format_value(desc)}, {format_value(desc_en)}, "
f"{format_value(customs)}, {format_value(american)}, {format_value(ace)}, {format_value(oma)}, 1, 1)"
for code, desc, desc_en, customs, american, ace, oma in units_of_measure_seed
])
val_uom = ", ".join(
[
f"({format_value(code)}, {format_value(desc)}, {format_value(desc_en)}, "
f"{format_value(customs)}, {format_value(american)}, {format_value(ace)}, {format_value(oma)}, 1, 1)"
for code, desc, desc_en, customs, american, ace, oma in units_of_measure_seed
]
)
op.execute("ALTER TABLE a76.units_of_measure DISABLE TRIGGER ALL;")
op.execute(f"""
op.execute(
f"""
INSERT INTO a76.units_of_measure
(code, description, description_en, customs_code, american_code, ace_code, oma_code, tenant_id, company_id)
VALUES {val_uom}
ON CONFLICT (code, tenant_id, company_id) DO NOTHING;
""")
"""
)
op.execute("ALTER TABLE a76.units_of_measure ENABLE TRIGGER ALL;")
# --- SEEDS CORE (Permissions) ---
# Combinar todas las seeds de permisos
all_permissions = seed_invoices + seed_user + seed_report + seed_roles
values_permissions = ", ".join([
f"({format_value(code)}, {format_value(desc)}, {format_value(module)}, {format_value(action)})"
for code, desc, module, action in all_permissions
])
values_permissions = ", ".join(
[
f"({format_value(code)}, {format_value(desc)}, {format_value(module)}, {format_value(action)})"
for code, desc, module, action in all_permissions
]
)
if values_permissions:
op.execute(f"""
op.execute(
f"""
INSERT INTO core.permissions (code, description, module, action)
VALUES {values_permissions}
ON CONFLICT (code) DO NOTHING;
""")
"""
)
# --- SEEDS PUBLIC (States) ---
values_states = ", ".join(
[
f"('{m3_key}', '{description.replace(chr(39), chr(39)*2)}', {format_value(mex_key)})"
for m3_key, description, mex_key in states_seed
]
)
op.execute(
f"""
INSERT INTO public.states (m3_key, description, mex_key)
VALUES {values_states}
ON CONFLICT (m3_key, description) DO NOTHING;
"""
)
def downgrade() -> None:
@@ -427,4 +491,4 @@ def downgrade() -> None:
op.drop_table("unit_of_measure_customs", schema="a76")
op.drop_table("unit_of_measure_american", schema="a76")
op.drop_table("unit_of_measure_oma", schema="a76")
op.drop_table("unit_of_measure_ace", schema="a76")
op.drop_table("unit_of_measure_ace", schema="a76")