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

feat: Relocate units of measure and historical tariff fractions seeding to company creation and add Celery schedule files to gitignore.

Browse Source
This commit is contained in:
Galindo97
2026-03-03 08:55:42 -07:00
parent 7c6376445f
commit 5bbb4ac9b9
3 changed files with 71 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.gitignore vendored
View File

@@ -67,3 +67,8 @@ postgres-data/
backend/uploads/
docker-compose.yml
.mypy_cache/
# Celery
backend/celerybeat-schedule
celerybeat-schedule
celerybeat-schedule.*

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

@@ -73,9 +73,6 @@ from api.v1.modules.a76.general_catalogs.units_of_measure.seed_adua import (
from api.v1.modules.a76.general_catalogs.fractions.tariff_fractions.seed import (
seed as tariff_fractions_seed,
)
from api.v1.modules.a76.general_catalogs.fractions.historical_tariff_fractions.seed import (
seed as historical_tariff_fractions_seed,
)
from api.v1.modules.public.reference_data.trailer_types.seed import (
seed as trailer_types_seed,
)
@@ -429,25 +426,7 @@ def upgrade() -> None:
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
]
)
op.execute("ALTER TABLE a76.units_of_measure DISABLE TRIGGER ALL;")
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;")
# TABLA MAESTRA UOM se genera ahora al crear una empresa
# --- SEEDS CORE (Permissions) ---
@@ -501,45 +480,7 @@ def upgrade() -> None:
"""
)
def format_bool(val):
"""Convert boolean string to SQL boolean."""
if val is None or str(val).strip() == "" or str(val).upper() == "NONE":
return "NULL"
return "TRUE" if str(val).upper() == "TRUE" else "FALSE"
def format_timestamp(val):
"""Format timestamp for PostgreSQL."""
if val is None or str(val).strip() == "" or str(val).upper() == "NONE":
return "NULL"
# El valor ya viene en formato 'YYYY-MM-DD HH:MM:SS'
return f"'{str(val)}'"
values_historical_fractions = ", ".join(
[
f"({format_value(historical_fraction)}, {format_value(nico)},{format_value(unit_measure)}, {format_value(country)}, "
f"{format_value(fraction_type)}, {format_value(sector)}, {format_value(import_tax)}, "
f"{format_value(export_tax)}, {format_timestamp(pub_date)}, {format_bool(is_immex)}, "
f"{format_bool(normal_temp)}, {format_bool(services_temp)}, {format_bool(certified_temp)}, "
f"{format_bool(by_log)}, {format_timestamp(end_date)}, "
f"1, 1)" # tenant_id=1, company_id=1
for historical_fraction, nico, unit_measure, country, fraction_type, sector, import_tax, export_tax, pub_date, is_immex, normal_temp, services_temp, certified_temp, by_log, end_date in historical_tariff_fractions_seed
]
)
if values_historical_fractions:
op.execute("SET session_replication_role = replica;")
op.execute(
f"""
INSERT INTO a76.historical_tariff_fractions
(historical_fraction, nico, unit_of_measure_code, country, fraction_type, sector,
import_tax_rate, export_tax_rate, publication_date, is_immex,
normal_temporality, services_temporality, certified_temporality, by_log, end_date,
tenant_id, company_id)
VALUES {values_historical_fractions}
ON CONFLICT DO NOTHING;
"""
)
op.execute("SET session_replication_role = DEFAULT;")
# Historical Fractions se generan ahora al crear una empresa
def downgrade() -> None:

65
backend/api/v1/modules/a76/general_catalogs/company/service.py
View File

@@ -12,7 +12,10 @@ from sqlalchemy.orm import Session
from .dto import CompanyCreateDTO, CompanyResponseDTO, CompanyUpdateDTO
from .models import Company
from ...audit_log.services.service import AuditService
from ..units_of_measure.seed import seed as units_of_measure_seed
from ..fractions.historical_tariff_fractions.seed import seed as historical_tariff_fractions_seed
from core.context import get_user_context
from sqlalchemy import text
logger = logging.getLogger(__name__)
@@ -381,7 +384,10 @@ class CompanyService:
if addr_ind2:
self.db.add(CompanyAddress(**addr_ind2, address_type='industrial2', company_id=db_company.id))
# 7. Commit
# 7. Seed company data (tenant/company dependent)
self._seed_company_data(self.db, tenant_id, db_company.id)
# 8. Commit
self.db.commit()
self.db.refresh(db_company)
@@ -698,6 +704,63 @@ class CompanyService:
# Custom methods
def _seed_company_data(self, db: Session, tenant_id: int, company_id: int):
"""Seeds tenant/company dependent data for a new company"""
def format_value(val):
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)}'"
# 1. Units of Measure
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)}, {tenant_id}, {company_id})"
for code, desc, desc_en, customs, american, ace, oma in units_of_measure_seed
]
)
db.execute(text("ALTER TABLE a76.units_of_measure DISABLE TRIGGER ALL;"))
db.execute(text(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;"))
db.execute(text("ALTER TABLE a76.units_of_measure ENABLE TRIGGER ALL;"))
# 2. Historical Tariff Fractions
def format_bool(val):
if val is None or str(val).strip() == "" or str(val).upper() == "NONE":
return "NULL"
return "TRUE" if str(val).upper() == "TRUE" else "FALSE"
def format_timestamp(val):
if val is None or str(val).strip() == "" or str(val).upper() == "NONE":
return "NULL"
return f"'{str(val)}'"
values_historical = ", ".join(
[
f"({format_value(historical_fraction)}, {format_value(nico)}, {format_value(unit_measure)}, {format_value(country)}, "
f"{format_value(fraction_type)}, {format_value(sector)}, {format_value(import_tax)}, "
f"{format_value(export_tax)}, {format_timestamp(pub_date)}, {format_bool(is_immex)}, "
f"{format_bool(normal_temp)}, {format_bool(services_temp)}, {format_bool(certified_temp)}, "
f"{format_bool(by_log)}, {format_timestamp(end_date)}, "
f"{tenant_id}, {company_id})"
for (historical_fraction, nico, unit_measure, country, fraction_type, sector, import_tax,
export_tax, pub_date, is_immex, normal_temp, services_temp, certified_temp, by_log, end_date) in historical_tariff_fractions_seed
]
)
if values_historical:
db.execute(text("SET session_replication_role = replica;"))
db.execute(text(f"""
INSERT INTO a76.historical_tariff_fractions
(historical_fraction, nico, unit_of_measure_code, country, fraction_type, sector,
import_tax_rate, export_tax_rate, publication_date, is_immex,
normal_temporality, services_temporality, certified_temporality, by_log, end_date,
tenant_id, company_id)
VALUES {values_historical}
ON CONFLICT DO NOTHING;
"""))
db.execute(text("SET session_replication_role = DEFAULT;"))
def get_companies_by_tenant(self, tenant_id: int) -> List[Company]:
"""Get all companies for a tenant"""
return (