Refactor models and services in A76 module
- Updated GBultoService to use models.Package instead of models.GBulto. - Refactored Part model to use SQLAlchemy 2.0 style with Mapped and mapped_column. - Added timestamps (created_at, updated_at, deleted_at) to various pedimento models. - Improved relationships and foreign key constraints in pedimento models. - Updated PermissionRuleOct and Seal models to use Mapped and mapped_column. - Changed DTO configuration from orm_mode to from_attributes for better compatibility. - Removed obsolete models (models.py and models_ped.py) from the repository.
This commit is contained in:
@@ -32,4 +32,4 @@ class GBultoResponseDTO(GBultoBaseDTO):
|
||||
UPDATED_AT: Optional[str]
|
||||
|
||||
class Config:
|
||||
orm_mode = True
|
||||
from_attributes = True
|
||||
@@ -1,22 +1,35 @@
|
||||
from sqlalchemy import Column, String, DECIMAL, DateTime, ForeignKey
|
||||
from sqlalchemy.orm import declarative_base
|
||||
from typing import Optional
|
||||
from datetime import datetime
|
||||
from decimal import Decimal
|
||||
from sqlalchemy import DateTime, Integer, String, DECIMAL, ForeignKey, PrimaryKeyConstraint, ForeignKeyConstraint, UniqueConstraint
|
||||
from sqlalchemy.sql import func
|
||||
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
from core.database import Base
|
||||
|
||||
class GBulto(Base):
|
||||
__tablename__ = "gbultos"
|
||||
__table_args__ = {"schema": "a76"}
|
||||
|
||||
CLAVE = Column(String(5), primary_key=True, nullable=False)
|
||||
DESCRIPTION = Column(String(40), nullable=True)
|
||||
DESCRIPTIONI = Column(String(40), nullable=True)
|
||||
WEIGHT_UNIT = Column(DECIMAL(19, 8), nullable=True)
|
||||
PLURALS = Column(String(4), nullable=True)
|
||||
PLURAL_IN = Column(String(4), nullable=True)
|
||||
CODE_ACE = Column(String(4), nullable=True)
|
||||
CODE_AAMEX = Column(String(9), nullable=True)
|
||||
CREATED_AT = Column(DateTime, server_default=func.now(), nullable=False)
|
||||
UPDATED_AT = Column(DateTime, server_default=func.now(), onupdate=func.now(), nullable=False)
|
||||
tenant_id = Column(String, ForeignKey("a76.tenants.id"), nullable=False)
|
||||
class Package(Base):
|
||||
__tablename__ = "packages" # GBultos
|
||||
__table_args__ = (
|
||||
PrimaryKeyConstraint('id', name='packages_pkey'),
|
||||
UniqueConstraint('key', name='packages_key_ukey'),
|
||||
ForeignKeyConstraint(['tenant_id'], ['a76.tenants.id'], name='fk_packages_tenant'),
|
||||
{"schema": "a76"}
|
||||
)
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
||||
tenant_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True)
|
||||
key: Mapped[str] = mapped_column(String(5))
|
||||
description_es: Mapped[Optional[str]] = mapped_column(String(40))
|
||||
description_en: Mapped[Optional[str]] = mapped_column(String(40))
|
||||
weight_unit: Mapped[Optional[Decimal]] = mapped_column(DECIMAL(19, 8))
|
||||
plurals: Mapped[Optional[str]] = mapped_column(String(4))
|
||||
plural_in: Mapped[Optional[str]] = mapped_column(String(4))
|
||||
code_ace: Mapped[Optional[str]] = mapped_column(String(4))
|
||||
code_aamex: Mapped[Optional[str]] = mapped_column(String(9))
|
||||
|
||||
# Timestamps
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=func.now())
|
||||
updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=func.now(), onupdate=func.now())
|
||||
deleted_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
|
||||
|
||||
|
||||
|
||||
@@ -4,11 +4,11 @@ from typing import List
|
||||
|
||||
from core.database import get_core_db
|
||||
from core.security import get_current_user
|
||||
from .models import GBulto
|
||||
from .models import Package
|
||||
from .dto import GBultoCreateDTO, GBultoUpdateDTO, GBultoResponseDTO
|
||||
from .services import GBultoService
|
||||
|
||||
router = APIRouter(prefix="/gbultos", tags=["GBultos"])
|
||||
router = APIRouter(prefix="/bultos", tags=["GBultos"])
|
||||
|
||||
|
||||
@router.get("/", response_model=List[GBultoResponseDTO])
|
||||
@@ -21,7 +21,7 @@ async def list_bultos(
|
||||
"""
|
||||
List all GBultos with pagination.
|
||||
"""
|
||||
return db.query(GBulto).offset(skip).limit(limit).all()
|
||||
return db.query(Package).offset(skip).limit(limit).all()
|
||||
|
||||
|
||||
@router.get("/{code}", response_model=GBultoResponseDTO)
|
||||
@@ -31,11 +31,11 @@ async def read_bulto(
|
||||
current_user: dict = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
Get a specific GBulto by its CODE.
|
||||
Get a specific Package by its CODE.
|
||||
"""
|
||||
bulto = GBultoService.get_bulto_by_code(db, code)
|
||||
if not bulto:
|
||||
raise HTTPException(status_code=404, detail="GBulto not found")
|
||||
raise HTTPException(status_code=404, detail="Package not found")
|
||||
return bulto
|
||||
|
||||
|
||||
@@ -46,7 +46,7 @@ async def create_gbulto(
|
||||
current_user: dict = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
Create a new GBulto.
|
||||
Create a new Package.
|
||||
"""
|
||||
return GBultoService.create_gbulto(db, bulto_data)
|
||||
|
||||
@@ -59,11 +59,11 @@ async def update_bulto(
|
||||
current_user: dict = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
Update an existing GBulto.
|
||||
Update an existing Package.
|
||||
"""
|
||||
bulto = GBultoService.update_bulto(db, code, bulto_data)
|
||||
if not bulto:
|
||||
raise HTTPException(status_code=404, detail="GBulto not found")
|
||||
raise HTTPException(status_code=404, detail="Package not found")
|
||||
return bulto
|
||||
|
||||
|
||||
@@ -74,8 +74,8 @@ async def delete_bulto(
|
||||
current_user: dict = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
Delete a GBulto by its CODE.
|
||||
Delete a Package by its CODE.
|
||||
"""
|
||||
bulto = GBultoService.delete_bulto(db, code)
|
||||
if not bulto:
|
||||
raise HTTPException(status_code=404, detail="GBulto not found")
|
||||
raise HTTPException(status_code=404, detail="Package not found")
|
||||
@@ -8,11 +8,11 @@ class GBultoService:
|
||||
|
||||
@staticmethod
|
||||
def get_bulto_by_code(db: Session, code: str):
|
||||
return db.query(models.GBulto).filter(models.GBulto.CODE == code).first()
|
||||
return db.query(models.Package).filter(models.Package.CODE == code).first()
|
||||
|
||||
@staticmethod
|
||||
def create_gbulto(db: Session, gbulto_data: dto.GBultoCreateDTO):
|
||||
new_gbulto = models.GBulto(**gbulto_data.dict())
|
||||
new_gbulto = models.Package(**gbulto_data.dict())
|
||||
db.add(new_gbulto)
|
||||
db.commit()
|
||||
db.refresh(new_gbulto)
|
||||
|
||||
Reference in New Issue
Block a user