Merge pull request 'Refactorización de nombres de módulos y actualizaciones:' (#9) from catalogos_g into development

Reviewed-on: ADUANASOFT/anexo76#9
This commit is contained in:
2025-11-07 16:10:19 +00:00
28 changed files with 797 additions and 1 deletions

View File

@@ -34,6 +34,7 @@ class ClientProvider(Base):
incoterm = Column(String(19), nullable=True)
is_national_provider = Column(String(2), nullable=True)
enabled_disabled = Column(SmallInteger, nullable=True)
tenant_id = Column(String, ForeignKey("a76.tenants.id"), nullable=False)
# Relationships
address = relationship("GClientProviderAddress", back_populates="client_provider", uselist=False, cascade="all, delete-orphan")

View File

@@ -1,7 +1,7 @@
"""
Modelos ORM para gestión de empresa
"""
from sqlalchemy import Column, Integer, String, DateTime, Boolean, Text, SmallInteger
from sqlalchemy import Column, Integer, String, DateTime, Boolean, Text, SmallInteger, ForeignKey
from sqlalchemy.sql import func
from core.database import Base
import enum
@@ -65,5 +65,7 @@ class GCompany(Base):
# Timestamps
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
updated_at = Column(DateTime(timezone=True), onupdate=func.now(), nullable=True)
tenant_id = Column(String, ForeignKey("a76.tenants.id"), nullable=False)

View File

@@ -0,0 +1,18 @@
"""
DTOs for CountryRuleOct.
"""
from pydantic import BaseModel
class CountryRuleOctBaseDTO(BaseModel):
permission: str
line: int
fraction: str
country_code: str
class CountryRuleOctCreateDTO(CountryRuleOctBaseDTO):
pass
class CountryRuleOctResponseDTO(CountryRuleOctBaseDTO):
class Config:
orm_mode = True

View File

@@ -0,0 +1,12 @@
from sqlalchemy import Column, String, Integer, ForeignKey
from core.database import Base
class CountryRuleOct(Base):
__tablename__ = "country_rule_oct"
__table_args__ = {"schema": "a76"}
permission = Column(String(20), ForeignKey("a76.GFracROctava.permission", ondelete="CASCADE"), primary_key=True, nullable=False)
line = Column(Integer, ForeignKey("a76.GFracROctava.line", ondelete="CASCADE"), primary_key=True, nullable=False)
fraction = Column(String(10), ForeignKey("a76.GFracROctava.fraction", ondelete="CASCADE"), primary_key=True, nullable=False)
country_code = Column(String(3), primary_key=True, nullable=False)
tenant_id = Column(String, ForeignKey("a76.tenants.id"), nullable=False)

View File

@@ -0,0 +1,68 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from core.database import get_core_db
from core.security import get_current_user
from .dto import CountryRuleOctCreateDTO, CountryRuleOctResponseDTO
from .services import CountryRuleOctService
router = APIRouter(prefix="/country-rule-oct", tags=["CountryRuleOct"])
@router.get("/", response_model=List[CountryRuleOctResponseDTO])
async def list_countries(
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
List all CountryRuleOct entries.
"""
return db.query(CountryRuleOctService).all()
@router.get("/{permission}/{line}/{fraction}/{country_code}", response_model=CountryRuleOctResponseDTO)
async def read_country_rule(
permission: str,
line: int,
fraction: str,
country_code: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Get a specific CountryRuleOct by its composite key.
"""
country = CountryRuleOctService.get_country_by_keys(db, permission, line, fraction, country_code)
if not country:
raise HTTPException(status_code=404, detail="CountryRuleOct not found")
return country
@router.post("/", response_model=CountryRuleOctResponseDTO, status_code=status.HTTP_201_CREATED)
async def create_country_rule(
country_data: CountryRuleOctCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Create a new CountryRuleOct entry.
"""
return CountryRuleOctService.create_country_rule(db, country_data)
@router.delete("/{permission}/{line}/{fraction}/{country_code}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_country_rule(
permission: str,
line: int,
fraction: str,
country_code: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Delete a CountryRuleOct by its composite key.
"""
country = CountryRuleOctService.delete_country_rule(db, permission, line, fraction, country_code)
if not country:
raise HTTPException(status_code=404, detail="CountryRuleOct not found")

View File

@@ -0,0 +1,32 @@
"""
Service layer for CountryRuleOct.
"""
from sqlalchemy.orm import Session
from . import models, dto
class CountryRuleOctService:
@staticmethod
def get_country_by_keys(db: Session, permission: str, line: int, fraction: str, country_code: str):
return db.query(models.CountryRuleOct).filter(
models.CountryRuleOct.permission == permission,
models.CountryRuleOct.line == line,
models.CountryRuleOct.fraction == fraction,
models.CountryRuleOct.country_code == country_code
).first()
@staticmethod
def create_country_rule(db: Session, country_data: dto.CountryRuleOctCreateDTO):
new_country = models.CountryRuleOct(**country_data.dict())
db.add(new_country)
db.commit()
db.refresh(new_country)
return new_country
@staticmethod
def delete_country_rule(db: Session, permission: str, line: int, fraction: str, country_code: str):
country = CountryRuleOctService.get_country_by_keys(db, permission, line, fraction, country_code)
if country:
db.delete(country)
db.commit()
return country

View File

@@ -0,0 +1,15 @@
from pydantic import BaseModel
from typing import Optional
class ExchangeRateBaseDTO(BaseModel):
date: int
value: Optional[float]
local_currency: Optional[str]
foreign_currency: Optional[str]
class ExchangeRateCreateDTO(ExchangeRateBaseDTO):
pass
class ExchangeRateResponseDTO(ExchangeRateBaseDTO):
class Config:
orm_mode = True

View File

@@ -0,0 +1,11 @@
from sqlalchemy import Column, Integer, String, DECIMAL
from core.database import Base
class ExchangeRate(Base):
__tablename__ = "exchange_rate"
__table_args__ = {"schema": "a76"}
date = Column(Integer, primary_key=True, nullable=False)
value = Column(DECIMAL(13, 6), nullable=True)
local_currency = Column(String(7), nullable=True)
foreign_currency = Column(String(7), nullable=True)

View File

@@ -0,0 +1,62 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from core.database import get_core_db
from core.security import get_current_user
from .dto import ExchangeRateCreateDTO, ExchangeRateResponseDTO
from .services import ExchangeRateService
router = APIRouter(prefix="/exchange-rate", tags=["ExchangeRate"])
@router.get("/", response_model=List[ExchangeRateResponseDTO])
async def list_exchange_rates(
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
List all ExchangeRate entries.
"""
return db.query(ExchangeRateService).all()
@router.get("/{date}", response_model=ExchangeRateResponseDTO)
async def read_exchange_rate(
date: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Get a specific ExchangeRate by its date.
"""
exchange_rate = ExchangeRateService.get_exchange_rate_by_date(db, date)
if not exchange_rate:
raise HTTPException(status_code=404, detail="ExchangeRate not found")
return exchange_rate
@router.post("/", response_model=ExchangeRateResponseDTO, status_code=status.HTTP_201_CREATED)
async def create_exchange_rate(
exchange_rate_data: ExchangeRateCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Create a new ExchangeRate entry.
"""
return ExchangeRateService.create_exchange_rate(db, exchange_rate_data)
@router.delete("/{date}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_exchange_rate(
date: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Delete an ExchangeRate by its date.
"""
exchange_rate = ExchangeRateService.delete_exchange_rate(db, date)
if not exchange_rate:
raise HTTPException(status_code=404, detail="ExchangeRate not found")

View File

@@ -0,0 +1,23 @@
from sqlalchemy.orm import Session
from . import models, dto
class ExchangeRateService:
@staticmethod
def get_exchange_rate_by_date(db: Session, date: int):
return db.query(models.ExchangeRate).filter(models.ExchangeRate.date == date).first()
@staticmethod
def create_exchange_rate(db: Session, exchange_rate_data: dto.ExchangeRateCreateDTO):
new_exchange_rate = models.ExchangeRate(**exchange_rate_data.dict())
db.add(new_exchange_rate)
db.commit()
db.refresh(new_exchange_rate)
return new_exchange_rate
@staticmethod
def delete_exchange_rate(db: Session, date: int):
exchange_rate = ExchangeRateService.get_exchange_rate_by_date(db, date)
if exchange_rate:
db.delete(exchange_rate)
db.commit()
return exchange_rate

View File

@@ -0,0 +1,24 @@
"""
DTOs for GFracROctava.
"""
from pydantic import BaseModel
from typing import Optional
class GFracROctavaBaseDTO(BaseModel):
PERMISSION: str
LINE: int
FRACTION: str
QUOTA_AMOUNT: Optional[float]
USED_AMOUNT: Optional[float]
QUOTA_VALUE: Optional[float]
USED_VALUE: Optional[float]
UNIT_COST_ME: Optional[float]
UNIT_MEASURE: Optional[str]
class GFracROctavaCreateDTO(GFracROctavaBaseDTO):
pass
class GFracROctavaResponseDTO(GFracROctavaBaseDTO):
class Config:
orm_mode = True

View File

@@ -0,0 +1,11 @@
from sqlalchemy import Column, String, Integer, ForeignKey
from core.database import Base
class FractionRuleOctave(Base):
__tablename__ = "fraction_rule_octave"
__table_args__ = {"schema": "a76"}
permission = Column(String(20), primary_key=True, nullable=False)
line = Column(Integer, nullable=False)
fraction = Column(String(10), nullable=False)
tenant_id = Column(String, ForeignKey("a76.tenants.id"), nullable=False)

View File

@@ -0,0 +1,66 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from core.database import get_core_db
from core.security import get_current_user
from .dto import GFracROctavaCreateDTO, GFracROctavaResponseDTO
from .services import GFracROctavaService
router = APIRouter(prefix="/gfracroctava", tags=["GFracROctava"])
@router.get("/", response_model=List[GFracROctavaResponseDTO])
async def list_fractions(
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
List all GFracROctava entries.
"""
return db.query(GFracROctavaService).all()
@router.get("/{permission}/{line}/{fraction}", response_model=GFracROctavaResponseDTO)
async def read_fraction(
permission: str,
line: int,
fraction: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Get a specific GFracROctava by its composite key.
"""
frac = GFracROctavaService.get_fraction_by_permission_line(db, permission, line, fraction)
if not frac:
raise HTTPException(status_code=404, detail="GFracROctava not found")
return frac
@router.post("/", response_model=GFracROctavaResponseDTO, status_code=status.HTTP_201_CREATED)
async def create_frac(
frac_data: GFracROctavaCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Create a new GFracROctava entry.
"""
return GFracROctavaService.create_frac(db, frac_data)
@router.delete("/{permission}/{line}/{fraction}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_fraction(
permission: str,
line: int,
fraction: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Delete a GFracROctava by its composite key.
"""
frac = GFracROctavaService.delete_fraction(db, permission, line, fraction)
if not frac:
raise HTTPException(status_code=404, detail="GFracROctava not found")

View File

@@ -0,0 +1,31 @@
from sqlalchemy.orm import Session
from . import models, dto
"""
Service layer for GFracROctava.
"""
class GFracROctavaService:
@staticmethod
def get_fraction_by_permission_line(db: Session, permission: str, line: int, fraction: str):
return db.query(models.GFracROctava).filter(
models.GFracROctava.PERMISSION == permission,
models.GFracROctava.LINE == line,
models.GFracROctava.FRACTION == fraction
).first()
@staticmethod
def create_frac(db: Session, frac_data: dto.GFracROctavaCreateDTO):
new_frac = models.GFracROctava(**frac_data.dict())
db.add(new_frac)
db.commit()
db.refresh(new_frac)
return new_frac
@staticmethod
def delete_fraction(db: Session, permission: str, line: int, fraction: str):
frac = GFracROctavaService.get_fraction_by_permission_line(db, permission, line, fraction)
if frac:
db.delete(frac)
db.commit()
return frac

View File

@@ -0,0 +1,35 @@
"""
DTOs for GBultos.
"""
from pydantic import BaseModel
from typing import Optional
class GBultoBaseDTO(BaseModel):
CODE: str
DESCRIPTION: Optional[str]
DESCRIPTIONI: Optional[str]
WEIGHT_UNIT: Optional[float]
PLURALS: Optional[str]
PLURAL_IN: Optional[str]
CODE_ACE: Optional[str]
CODE_AAMEX: Optional[str]
class GBultoCreateDTO(GBultoBaseDTO):
pass
class GBultoUpdateDTO(BaseModel):
DESCRIPTION: Optional[str]
DESCRIPTIONI: Optional[str]
WEIGHT_UNIT: Optional[float]
PLURALS: Optional[str]
PLURAL_IN: Optional[str]
CODE_ACE: Optional[str]
CODE_AAMEX: Optional[str]
class GBultoResponseDTO(GBultoBaseDTO):
CREATED_AT: Optional[str]
UPDATED_AT: Optional[str]
class Config:
orm_mode = True

View File

@@ -0,0 +1,22 @@
from sqlalchemy import Column, String, DECIMAL, DateTime, ForeignKey
from sqlalchemy.orm import declarative_base
from sqlalchemy.sql import func
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)

View File

@@ -0,0 +1,81 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from core.database import get_core_db
from core.security import get_current_user
from .models import GBulto
from .dto import GBultoCreateDTO, GBultoUpdateDTO, GBultoResponseDTO
from .services import GBultoService
router = APIRouter(prefix="/gbultos", tags=["GBultos"])
@router.get("/", response_model=List[GBultoResponseDTO])
async def list_bultos(
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
List all GBultos with pagination.
"""
return db.query(GBulto).offset(skip).limit(limit).all()
@router.get("/{code}", response_model=GBultoResponseDTO)
async def read_bulto(
code: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Get a specific GBulto by its CODE.
"""
bulto = GBultoService.get_bulto_by_code(db, code)
if not bulto:
raise HTTPException(status_code=404, detail="GBulto not found")
return bulto
@router.post("/", response_model=GBultoResponseDTO, status_code=status.HTTP_201_CREATED)
async def create_gbulto(
bulto_data: GBultoCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Create a new GBulto.
"""
return GBultoService.create_gbulto(db, bulto_data)
@router.put("/{code}", response_model=GBultoResponseDTO)
async def update_bulto(
code: str,
bulto_data: GBultoUpdateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Update an existing GBulto.
"""
bulto = GBultoService.update_bulto(db, code, bulto_data)
if not bulto:
raise HTTPException(status_code=404, detail="GBulto not found")
return bulto
@router.delete("/{code}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_bulto(
code: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Delete a GBulto by its CODE.
"""
bulto = GBultoService.delete_bulto(db, code)
if not bulto:
raise HTTPException(status_code=404, detail="GBulto not found")

View File

@@ -0,0 +1,37 @@
from sqlalchemy.orm import Session
from . import models, dto
class GBultoService:
"""
Service layer for GBultos.
"""
@staticmethod
def get_bulto_by_code(db: Session, code: str):
return db.query(models.GBulto).filter(models.GBulto.CODE == code).first()
@staticmethod
def create_gbulto(db: Session, gbulto_data: dto.GBultoCreateDTO):
new_gbulto = models.GBulto(**gbulto_data.dict())
db.add(new_gbulto)
db.commit()
db.refresh(new_gbulto)
return new_gbulto
@staticmethod
def update_bulto(db: Session, code: str, bulto_data: dto.GBultoUpdateDTO):
bulto = GBultoService.get_bulto_by_code(db, code)
if bulto:
for key, value in bulto_data.dict(exclude_unset=True).items():
setattr(bulto, key, value)
db.commit()
db.refresh(bulto)
return bulto
@staticmethod
def delete_bulto(db: Session, code: str):
bulto = GBultoService.get_bulto_by_code(db, code)
if bulto:
db.delete(bulto)
db.commit()
return bulto

View File

@@ -82,6 +82,8 @@ class Part(Base):
back_populates="parts"
)
tenant_id = Column(String, ForeignKey("a76.tenants.id"), nullable=False)
def __repr__(self):
return f"<Part(client_key={self.client_key}, part_number='{self.part_number}', description='{self.description_spanish}')>"

View File

@@ -0,0 +1,16 @@
from pydantic import BaseModel
from typing import Optional
class PermissionRuleOctBaseDTO(BaseModel):
permission: str
start_date: Optional[int]
end_date: Optional[int]
sector: Optional[str]
system: Optional[str]
class PermissionRuleOctCreateDTO(PermissionRuleOctBaseDTO):
pass
class PermissionRuleOctResponseDTO(PermissionRuleOctBaseDTO):
class Config:
orm_mode = True

View File

@@ -0,0 +1,13 @@
from sqlalchemy import Column, String, Integer, ForeignKey
from core.database import Base
class PermissionRuleOct(Base):
__tablename__ = "permission_rule_oct"
__table_args__ = {"schema": "a76"}
permission = Column(String(20), primary_key=True, nullable=False)
start_date = Column(Integer, nullable=True)
end_date = Column(Integer, nullable=True)
sector = Column(String(8), nullable=True)
system = Column(String(5), nullable=True)
tenant_id = Column(String, ForeignKey("a76.tenants.id"), nullable=False)

View File

@@ -0,0 +1,62 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from core.database import get_core_db
from core.security import get_current_user
from .dto import PermissionRuleOctCreateDTO, PermissionRuleOctResponseDTO
from .services import PermissionRuleOctService
router = APIRouter(prefix="/permission-rule-oct", tags=["PermissionRuleOct"])
@router.get("/", response_model=List[PermissionRuleOctResponseDTO])
async def list_permissions(
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
List all PermissionRuleOct entries.
"""
return db.query(PermissionRuleOctService).all()
@router.get("/{permission}", response_model=PermissionRuleOctResponseDTO)
async def read_permission(
permission: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Get a specific PermissionRuleOct by its permission.
"""
permission = PermissionRuleOctService.get_permission_by_id(db, permission)
if not permission:
raise HTTPException(status_code=404, detail="PermissionRuleOct not found")
return permission
@router.post("/", response_model=PermissionRuleOctResponseDTO, status_code=status.HTTP_201_CREATED)
async def create_permission(
permission_data: PermissionRuleOctCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Create a new PermissionRuleOct entry.
"""
return PermissionRuleOctService.create_permission(db, permission_data)
@router.delete("/{permission}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_permission(
permission: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Delete a PermissionRuleOct by its permission.
"""
permission = PermissionRuleOctService.delete_permission(db, permission)
if not permission:
raise HTTPException(status_code=404, detail="PermissionRuleOct not found")

View File

@@ -0,0 +1,23 @@
from sqlalchemy.orm import Session
from . import models, dto
class PermissionRuleOctService:
@staticmethod
def get_permission_by_id(db: Session, permission: str):
return db.query(models.PermissionRuleOct).filter(models.PermissionRuleOct.permission == permission).first()
@staticmethod
def create_permission(db: Session, permission_data: dto.PermissionRuleOctCreateDTO):
new_permission = models.PermissionRuleOct(**permission_data.dict())
db.add(new_permission)
db.commit()
db.refresh(new_permission)
return new_permission
@staticmethod
def delete_permission(db: Session, permission: str):
permission = PermissionRuleOctService.get_permission_by_id(db, permission)
if permission:
db.delete(permission)
db.commit()
return permission

View File

@@ -13,6 +13,12 @@ from .client_and_provider import router as client_and_provider_router
from .company import router as company_router
from .classes import router as classes_router
from .parts import router as parts_router
from .permission_rule_oct.routes import router as permission_rule_oct_router
from .package.routes import router as package_router
from .seal.routes import router as seal_router
from .fraction_rule_octave.routes import router as fraction_rule_octave_router
from .country_rule_oct.routes import router as country_rule_oct_router
from .exchange_rate.routes import router as exchange_rate_router
# Router principal
router = APIRouter()
@@ -26,4 +32,10 @@ router.include_router(client_and_provider_router, prefix="/a76", tags=["a76 / cl
router.include_router(company_router, prefix="/a76", tags=["a76 / company"])
router.include_router(classes_router, prefix="/a76", tags=["a76 / classes"])
router.include_router(parts_router, prefix="/a76", tags=["a76 / parts"])
router.include_router(permission_rule_oct_router, prefix="/a76", tags=["a76 / PermissionRuleOct"])
router.include_router(package_router, prefix="/a76", tags=["a76 / Package"])
router.include_router(seal_router, prefix="/a76", tags=["a76 / Seal"])
router.include_router(fraction_rule_octave_router, prefix="/a76", tags=["a76 / FractionRuleOctave"])
router.include_router(country_rule_oct_router, prefix="/a76", tags=["a76 / CountryRuleOct"])
router.include_router(exchange_rate_router, prefix="/a76", tags=["a76 / ExchangeRate"])

View File

@@ -0,0 +1,15 @@
"""
DTOs for Seal.
"""
from pydantic import BaseModel
class SealBaseDTO(BaseModel):
seal: str
class SealCreateDTO(SealBaseDTO):
pass
class SealResponseDTO(SealBaseDTO):
class Config:
orm_mode = True

View File

@@ -0,0 +1,9 @@
from sqlalchemy import Column, String, ForeignKey
from core.database import Base
class Seal(Base):
__tablename__ = "seal"
__table_args__ = {"schema": "a76"}
seal = Column(String(15), primary_key=True, nullable=False)
tenant_id = Column(String, ForeignKey("a76.tenants.id"), nullable=False)

View File

@@ -0,0 +1,66 @@
"""
Routes for managing Seal entries.
"""
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from core.database import get_core_db
from core.security import get_current_user
from .dto import SealCreateDTO, SealResponseDTO
from .services import SealService
router = APIRouter(prefix="/seals", tags=["Seal"])
@router.get("/", response_model=List[SealResponseDTO])
async def list_seals(
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
List all Seal entries.
"""
return db.query(SealService).all()
@router.get("/{seal}", response_model=SealResponseDTO)
async def read_seal(
seal: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Get a specific Seal by its seal.
"""
seal = SealService.get_seal_by_id(db, seal)
if not seal:
raise HTTPException(status_code=404, detail="Seal not found")
return seal
@router.post("/", response_model=SealResponseDTO, status_code=status.HTTP_201_CREATED)
async def create_seal(
seal_data: SealCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Create a new Seal entry.
"""
return SealService.create_seal(db, seal_data)
@router.delete("/{seal}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_seal(
seal: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Delete a Seal by its seal.
"""
seal = SealService.delete_seal(db, seal)
if not seal:
raise HTTPException(status_code=404, detail="Seal not found")

View File

@@ -0,0 +1,27 @@
from sqlalchemy.orm import Session
from . import models, dto
"""
Service layer for Seal.
"""
class SealService:
@staticmethod
def get_seal_by_id(db: Session, seal: str):
return db.query(models.Seal).filter(models.Seal.seal == seal).first()
@staticmethod
def create_seal(db: Session, seal_data: dto.SealCreateDTO):
new_seal = models.Seal(**seal_data.dict())
db.add(new_seal)
db.commit()
db.refresh(new_seal)
return new_seal
@staticmethod
def delete_seal(db: Session, seal: str):
seal = SealService.get_seal_by_id(db, seal)
if seal:
db.delete(seal)
db.commit()
return seal