feat(company): add logo upload functionality and update logo field size

This commit is contained in:
2026-01-13 13:06:14 -06:00
parent c00107d921
commit 0a1b6cd1e0
6 changed files with 317 additions and 41 deletions

View File

@@ -57,7 +57,7 @@ class Company(Base, TimestampMixin):
position: Mapped[Optional[str]] = mapped_column(String(30))
# Configuración
logo: Mapped[Optional[str]] = mapped_column(String(255))
logo: Mapped[Optional[str]] = mapped_column(String(500))
has_express_line: Mapped[Optional[bool]] = mapped_column(Boolean)
order_format_type: Mapped[Optional[str]] = mapped_column(String(19))
previous_code: Mapped[Optional[int]] = mapped_column(SmallInteger)

View File

@@ -2,9 +2,12 @@
Rutas para gestión de empresa
"""
import os
import shutil
from typing import List, Optional
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi import APIRouter, Depends, HTTPException, status, UploadFile, File
from sqlalchemy.orm import Session
from core.database import get_core_db
@@ -14,9 +17,15 @@ from .dto import CompanyCreateDTO, CompanyResponseDTO, CompanyUpdateDTO
from .models import Company
from .service import CompanyService
# Configuración de directorios
UPLOAD_DIR = "uploads/companies"
ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
MAX_FILE_SIZE = 5 * 1024 * 1024 # 5MB
# Main router that includes base CRUD
router = APIRouter(prefix="/company")
@router.post(
"", # Se suma al prefix, queda POST /api/v1/a76/company
response_model=CompanyResponseDTO,
@@ -28,7 +37,7 @@ async def create_company(
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
tenant_id = current_user.get("tenant_id")
if not tenant_id:
raise HTTPException(
@@ -70,12 +79,12 @@ async def list_companies(
service = CompanyService(db)
items, total = service.get_all(
db,
tenant_id,
db,
tenant_id,
company_id=0, # Not used for companies
skip=skip,
skip=skip,
limit=page_size,
filters=filters if filters else None
filters=filters if filters else None,
)
total_pages = (total + page_size - 1) // page_size
@@ -286,9 +295,7 @@ async def update_company(
detail="Tenant ID not found in user data",
)
updated_company = CompanyService.update(
db, company_id, tenant_id, 0, data
)
updated_company = CompanyService.update(db, company_id, tenant_id, 0, data)
if not updated_company:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
@@ -323,4 +330,86 @@ async def delete_company(
detail="Company not found",
)
return None
return None
@router.post(
"/{company_id}/upload-logo",
response_model=dict,
summary="Upload company logo",
)
async def upload_company_logo(
company_id: int,
file: UploadFile = File(...),
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""Upload a logo for a company"""
tenant_id = current_user.get("tenant_id")
if not tenant_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Tenant ID not found in user data",
)
# Validar que la empresa existe
company = CompanyService.get_by_id(db, company_id, tenant_id, 0)
if not company:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Company not found",
)
# Validar extensión
file_ext = os.path.splitext(file.filename)[1].lower()
if file_ext not in ALLOWED_EXTENSIONS:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"File type not allowed. Allowed: {', '.join(ALLOWED_EXTENSIONS)}",
)
# Validar tamaño
content = await file.read()
if len(content) > MAX_FILE_SIZE:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"File too large. Maximum size: {MAX_FILE_SIZE / 1024 / 1024}MB",
)
# Crear directorio si no existe
os.makedirs(UPLOAD_DIR, exist_ok=True)
# Eliminar logo anterior si existe
if company.logo:
old_logo_path = company.logo
if os.path.exists(old_logo_path):
try:
os.remove(old_logo_path)
except Exception:
pass # No es crítico si falla
# Generar nombre único
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"company_{company_id}_{timestamp}{file_ext}"
file_path = os.path.join(UPLOAD_DIR, filename)
# Guardar archivo
try:
await file.seek(0)
with open(file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Error saving file: {str(e)}",
)
# Actualizar la empresa con la ruta del logo
update_data = CompanyUpdateDTO(logo=file_path)
updated_company = CompanyService.update(db, company_id, tenant_id, 0, update_data)
return {
"message": "Logo uploaded successfully",
"logo_path": file_path,
"company_id": company_id,
}