From 4ca738dc7f6ef28a0b2e06c112e580ae0abbdbac Mon Sep 17 00:00:00 2001 From: Samir Boulahtit Date: Mon, 1 Dec 2025 21:50:09 +0100 Subject: [PATCH] feat: implement company-based ownership architecture - Add database migration to make vendor.owner_user_id nullable - Update Vendor model to support company-based ownership (DEPRECATED vendor.owner_user_id) - Implement company_service with singleton pattern (consistent with vendor_service) - Create Company model with proper relationships to vendors and users - Add company exception classes for proper error handling - Refactor companies API to use singleton service pattern Architecture Change: - OLD: Each vendor has its own owner (vendor.owner_user_id) - NEW: Vendors belong to a company, company has one owner (company.owner_user_id) - This allows one company owner to manage multiple vendor brands Technical Details: - Company service uses singleton pattern (not factory) - Company service accepts db: Session as parameter (follows SVC-003) - Uses AuthManager for password hashing (consistent with admin_service) - Added _generate_temp_password() helper method --- ...make_vendor_owner_user_id_nullable_for_.py | 48 +++ app/api/v1/admin/companies.py | 297 ++++++++++++++++++ app/exceptions/company.py | 128 ++++++++ app/services/company_service.py | 266 ++++++++++++++++ models/database/company.py | 106 +++++++ models/database/vendor.py | 37 ++- models/schema/company.py | 155 +++++++++ 7 files changed, 1018 insertions(+), 19 deletions(-) create mode 100644 alembic/versions/5818330181a5_make_vendor_owner_user_id_nullable_for_.py create mode 100644 app/api/v1/admin/companies.py create mode 100644 app/exceptions/company.py create mode 100644 app/services/company_service.py create mode 100644 models/database/company.py create mode 100644 models/schema/company.py diff --git a/alembic/versions/5818330181a5_make_vendor_owner_user_id_nullable_for_.py b/alembic/versions/5818330181a5_make_vendor_owner_user_id_nullable_for_.py new file mode 100644 index 00000000..cb3a74bc --- /dev/null +++ b/alembic/versions/5818330181a5_make_vendor_owner_user_id_nullable_for_.py @@ -0,0 +1,48 @@ +"""make_vendor_owner_user_id_nullable_for_company_ownership + +Revision ID: 5818330181a5 +Revises: d0325d7c0f25 +Create Date: 2025-12-01 20:30:06.158027 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = '5818330181a5' +down_revision: Union[str, None] = 'd0325d7c0f25' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """ + Make vendor.owner_user_id nullable to support company-level ownership. + + Architecture Change: + - OLD: Each vendor has its own owner (vendor.owner_user_id) + - NEW: Vendors belong to a company, company has one owner (company.owner_user_id) + + This allows one company owner to manage multiple vendor brands. + """ + # Use batch operations for SQLite compatibility + with op.batch_alter_table('vendors', schema=None) as batch_op: + batch_op.alter_column('owner_user_id', + existing_type=sa.INTEGER(), + nullable=True) + + +def downgrade() -> None: + """ + Revert vendor.owner_user_id to non-nullable. + + WARNING: This will fail if there are vendors without owner_user_id! + """ + # Use batch operations for SQLite compatibility + with op.batch_alter_table('vendors', schema=None) as batch_op: + batch_op.alter_column('owner_user_id', + existing_type=sa.INTEGER(), + nullable=False) diff --git a/app/api/v1/admin/companies.py b/app/api/v1/admin/companies.py new file mode 100644 index 00000000..dc542197 --- /dev/null +++ b/app/api/v1/admin/companies.py @@ -0,0 +1,297 @@ +# app/api/v1/admin/companies.py +""" +Company management endpoints for admin. +""" + +import logging + +from fastapi import APIRouter, Body, Depends, Path, Query +from sqlalchemy.orm import Session + +from app.api.deps import get_current_admin_api +from app.core.database import get_db +from app.exceptions import CompanyHasVendorsException, ConfirmationRequiredException +from app.services.company_service import company_service +from models.database.user import User +from models.schema.company import ( + CompanyCreate, + CompanyCreateResponse, + CompanyDetailResponse, + CompanyListResponse, + CompanyResponse, + CompanyUpdate, +) + +router = APIRouter(prefix="/companies") +logger = logging.getLogger(__name__) + + +@router.post("", response_model=CompanyCreateResponse) +def create_company_with_owner( + company_data: CompanyCreate, + db: Session = Depends(get_db), + current_admin: User = Depends(get_current_admin_api), +): + """ + Create a new company with owner user account (Admin only). + + This endpoint: + 1. Creates a new company record + 2. Creates an owner user account with owner_email (if not exists) + 3. Returns credentials (temporary password shown ONCE if new user created) + + **Email Fields:** + - `owner_email`: Used for owner's login/authentication (stored in users.email) + - `contact_email`: Public business contact (stored in companies.contact_email) + + Returns company details with owner credentials. + """ + company, owner_user, temp_password = company_service.create_company_with_owner( + db, company_data + ) + + db.commit() + + return CompanyCreateResponse( + company=CompanyResponse( + id=company.id, + name=company.name, + description=company.description, + owner_user_id=company.owner_user_id, + contact_email=company.contact_email, + contact_phone=company.contact_phone, + website=company.website, + business_address=company.business_address, + tax_number=company.tax_number, + is_active=company.is_active, + is_verified=company.is_verified, + created_at=company.created_at.isoformat(), + updated_at=company.updated_at.isoformat(), + ), + owner_user_id=owner_user.id, + owner_username=owner_user.username, + owner_email=owner_user.email, + temporary_password=temp_password or "N/A (Existing user)", + login_url=f"http://localhost:8000/admin/login", + ) + + +@router.get("", response_model=CompanyListResponse) +def get_all_companies( + skip: int = Query(0, ge=0), + limit: int = Query(100, ge=1, le=1000), + search: str | None = Query(None, description="Search by company name"), + is_active: bool | None = Query(None), + is_verified: bool | None = Query(None), + db: Session = Depends(get_db), + current_admin: User = Depends(get_current_admin_api), +): + """Get all companies with filtering (Admin only).""" + companies, total = company_service.get_companies( + db, + skip=skip, + limit=limit, + search=search, + is_active=is_active, + is_verified=is_verified, + ) + + return CompanyListResponse( + companies=[ + CompanyResponse( + id=c.id, + name=c.name, + description=c.description, + owner_user_id=c.owner_user_id, + contact_email=c.contact_email, + contact_phone=c.contact_phone, + website=c.website, + business_address=c.business_address, + tax_number=c.tax_number, + is_active=c.is_active, + is_verified=c.is_verified, + created_at=c.created_at.isoformat(), + updated_at=c.updated_at.isoformat(), + ) + for c in companies + ], + total=total, + skip=skip, + limit=limit, + ) + + +@router.get("/{company_id}", response_model=CompanyDetailResponse) +def get_company_details( + company_id: int = Path(..., description="Company ID"), + db: Session = Depends(get_db), + current_admin: User = Depends(get_current_admin_api), +): + """ + Get detailed company information including vendor counts (Admin only). + """ + company = company_service.get_company_by_id(db, company_id) + + # Count vendors + vendor_count = len(company.vendors) + active_vendor_count = sum(1 for v in company.vendors if v.is_active) + + return CompanyDetailResponse( + id=company.id, + name=company.name, + description=company.description, + owner_user_id=company.owner_user_id, + contact_email=company.contact_email, + contact_phone=company.contact_phone, + website=company.website, + business_address=company.business_address, + tax_number=company.tax_number, + is_active=company.is_active, + is_verified=company.is_verified, + created_at=company.created_at.isoformat(), + updated_at=company.updated_at.isoformat(), + vendor_count=vendor_count, + active_vendor_count=active_vendor_count, + ) + + +@router.put("/{company_id}", response_model=CompanyResponse) +def update_company( + company_id: int = Path(..., description="Company ID"), + company_update: CompanyUpdate = Body(...), + db: Session = Depends(get_db), + current_admin: User = Depends(get_current_admin_api), +): + """ + Update company information (Admin only). + + **Can update:** + - Basic info: name, description + - Business contact: contact_email, contact_phone, website + - Business details: business_address, tax_number + - Status: is_active, is_verified + + **Cannot update:** + - `owner_user_id` (would require ownership transfer feature) + """ + company = company_service.update_company(db, company_id, company_update) + db.commit() + + return CompanyResponse( + id=company.id, + name=company.name, + description=company.description, + owner_user_id=company.owner_user_id, + contact_email=company.contact_email, + contact_phone=company.contact_phone, + website=company.website, + business_address=company.business_address, + tax_number=company.tax_number, + is_active=company.is_active, + is_verified=company.is_verified, + created_at=company.created_at.isoformat(), + updated_at=company.updated_at.isoformat(), + ) + + +@router.put("/{company_id}/verification", response_model=CompanyResponse) +def toggle_company_verification( + company_id: int = Path(..., description="Company ID"), + verification_data: dict = Body(..., example={"is_verified": True}), + db: Session = Depends(get_db), + current_admin: User = Depends(get_current_admin_api), +): + """ + Toggle company verification status (Admin only). + + Request body: { "is_verified": true/false } + """ + is_verified = verification_data.get("is_verified", False) + company = company_service.toggle_verification(db, company_id, is_verified) + db.commit() + + return CompanyResponse( + id=company.id, + name=company.name, + description=company.description, + owner_user_id=company.owner_user_id, + contact_email=company.contact_email, + contact_phone=company.contact_phone, + website=company.website, + business_address=company.business_address, + tax_number=company.tax_number, + is_active=company.is_active, + is_verified=company.is_verified, + created_at=company.created_at.isoformat(), + updated_at=company.updated_at.isoformat(), + ) + + +@router.put("/{company_id}/status", response_model=CompanyResponse) +def toggle_company_status( + company_id: int = Path(..., description="Company ID"), + status_data: dict = Body(..., example={"is_active": True}), + db: Session = Depends(get_db), + current_admin: User = Depends(get_current_admin_api), +): + """ + Toggle company active status (Admin only). + + Request body: { "is_active": true/false } + """ + is_active = status_data.get("is_active", True) + company = company_service.toggle_active(db, company_id, is_active) + db.commit() + + return CompanyResponse( + id=company.id, + name=company.name, + description=company.description, + owner_user_id=company.owner_user_id, + contact_email=company.contact_email, + contact_phone=company.contact_phone, + website=company.website, + business_address=company.business_address, + tax_number=company.tax_number, + is_active=company.is_active, + is_verified=company.is_verified, + created_at=company.created_at.isoformat(), + updated_at=company.updated_at.isoformat(), + ) + + +@router.delete("/{company_id}") +def delete_company( + company_id: int = Path(..., description="Company ID"), + confirm: bool = Query(False, description="Must be true to confirm deletion"), + db: Session = Depends(get_db), + current_admin: User = Depends(get_current_admin_api), +): + """ + Delete company and all associated vendors (Admin only). + + ⚠️ **WARNING: This is destructive and will delete:** + - Company account + - All vendors under this company + - All products under those vendors + - All orders, customers, team members + + Requires confirmation parameter: `confirm=true` + """ + if not confirm: + raise ConfirmationRequiredException( + operation="delete_company", + message="Deletion requires confirmation parameter: confirm=true", + ) + + # Get company to check vendor count + company = company_service.get_company_by_id(db, company_id) + vendor_count = len(company.vendors) + + if vendor_count > 0: + raise CompanyHasVendorsException(company_id, vendor_count) + + company_service.delete_company(db, company_id) + db.commit() + + return {"message": f"Company {company_id} deleted successfully"} diff --git a/app/exceptions/company.py b/app/exceptions/company.py new file mode 100644 index 00000000..491e4e14 --- /dev/null +++ b/app/exceptions/company.py @@ -0,0 +1,128 @@ +# app/exceptions/company.py +""" +Company management specific exceptions. +""" + +from typing import Any + +from .base import ( + AuthorizationException, + BusinessLogicException, + ConflictException, + ResourceNotFoundException, + ValidationException, +) + + +class CompanyNotFoundException(ResourceNotFoundException): + """Raised when a company is not found.""" + + def __init__(self, company_identifier: str | int, identifier_type: str = "id"): + if identifier_type.lower() == "id": + message = f"Company with ID '{company_identifier}' not found" + else: + message = f"Company with name '{company_identifier}' not found" + + super().__init__( + resource_type="Company", + identifier=str(company_identifier), + message=message, + error_code="COMPANY_NOT_FOUND", + ) + + +class CompanyAlreadyExistsException(ConflictException): + """Raised when trying to create a company that already exists.""" + + def __init__(self, company_name: str): + super().__init__( + message=f"Company with name '{company_name}' already exists", + error_code="COMPANY_ALREADY_EXISTS", + details={"company_name": company_name}, + ) + + +class CompanyNotActiveException(BusinessLogicException): + """Raised when trying to perform operations on inactive company.""" + + def __init__(self, company_id: int): + super().__init__( + message=f"Company with ID '{company_id}' is not active", + error_code="COMPANY_NOT_ACTIVE", + details={"company_id": company_id}, + ) + + +class CompanyNotVerifiedException(BusinessLogicException): + """Raised when trying to perform operations requiring verified company.""" + + def __init__(self, company_id: int): + super().__init__( + message=f"Company with ID '{company_id}' is not verified", + error_code="COMPANY_NOT_VERIFIED", + details={"company_id": company_id}, + ) + + +class UnauthorizedCompanyAccessException(AuthorizationException): + """Raised when user tries to access company they don't own.""" + + def __init__(self, company_id: int, user_id: int | None = None): + details = {"company_id": company_id} + if user_id: + details["user_id"] = user_id + + super().__init__( + message=f"Unauthorized access to company with ID '{company_id}'", + error_code="UNAUTHORIZED_COMPANY_ACCESS", + details=details, + ) + + +class InvalidCompanyDataException(ValidationException): + """Raised when company data is invalid or incomplete.""" + + def __init__( + self, + message: str = "Invalid company data", + field: str | None = None, + details: dict[str, Any] | None = None, + ): + super().__init__( + message=message, + field=field, + details=details, + ) + self.error_code = "INVALID_COMPANY_DATA" + + +class CompanyValidationException(ValidationException): + """Raised when company validation fails.""" + + def __init__( + self, + message: str = "Company validation failed", + field: str | None = None, + validation_errors: dict[str, str] | None = None, + ): + details = {} + if validation_errors: + details["validation_errors"] = validation_errors + + super().__init__( + message=message, + field=field, + details=details, + ) + self.error_code = "COMPANY_VALIDATION_FAILED" + + +class CompanyHasVendorsException(BusinessLogicException): + """Raised when trying to delete a company that still has active vendors.""" + + def __init__(self, company_id: int, vendor_count: int): + super().__init__( + message=f"Cannot delete company with ID '{company_id}' because it has {vendor_count} associated vendor(s)", + error_code="COMPANY_HAS_VENDORS", + details={"company_id": company_id, "vendor_count": vendor_count}, + ) diff --git a/app/services/company_service.py b/app/services/company_service.py new file mode 100644 index 00000000..17503608 --- /dev/null +++ b/app/services/company_service.py @@ -0,0 +1,266 @@ +# app/services/company_service.py +""" +Company service for managing company operations. + +This service handles CRUD operations for companies and company-vendor relationships. +""" + +import logging +import secrets +import string +from typing import List, Optional + +from sqlalchemy import func, select +from sqlalchemy.orm import Session, joinedload + +from app.exceptions import CompanyNotFoundException +from models.database.company import Company +from models.database.user import User +from models.schema.company import CompanyCreate, CompanyUpdate + +logger = logging.getLogger(__name__) + + +class CompanyService: + """Service for managing companies.""" + + def __init__(self): + """Initialize company service.""" + pass + + def create_company_with_owner( + self, db: Session, company_data: CompanyCreate + ) -> tuple[Company, User, str]: + """ + Create a new company with an owner user account. + + Args: + db: Database session + company_data: Company creation data + + Returns: + Tuple of (company, owner_user, temporary_password) + """ + # Import AuthManager for password hashing (same pattern as admin_service) + from middleware.auth import AuthManager + + auth_manager = AuthManager() + + # Check if owner email already exists + existing_user = db.execute( + select(User).where(User.email == company_data.owner_email) + ).scalar_one_or_none() + + if existing_user: + # Use existing user as owner + owner_user = existing_user + temp_password = None + logger.info(f"Using existing user {owner_user.email} as company owner") + else: + # Generate temporary password for owner + temp_password = self._generate_temp_password() + + # Create new owner user + owner_user = User( + username=company_data.owner_email.split("@")[0], + email=company_data.owner_email, + hashed_password=auth_manager.hash_password(temp_password), + role="user", + is_active=True, + is_email_verified=True, + ) + db.add(owner_user) + db.flush() # Get owner_user.id + logger.info(f"Created new owner user: {owner_user.email}") + + # Create company + company = Company( + name=company_data.name, + description=company_data.description, + owner_user_id=owner_user.id, + contact_email=company_data.contact_email, + contact_phone=company_data.contact_phone, + website=company_data.website, + business_address=company_data.business_address, + tax_number=company_data.tax_number, + is_active=True, + is_verified=False, + ) + db.add(company) + db.flush() + logger.info(f"Created company: {company.name} (ID: {company.id})") + + return company, owner_user, temp_password + + def get_company_by_id(self, db: Session, company_id: int) -> Company: + """ + Get company by ID. + + Args: + db: Database session + company_id: Company ID + + Returns: + Company object + + Raises: + CompanyNotFoundException: If company not found + """ + company = db.execute( + select(Company) + .where(Company.id == company_id) + .options(joinedload(Company.vendors)) + ).scalar_one_or_none() + + if not company: + raise CompanyNotFoundException(company_id) + + return company + + def get_companies( + self, + db: Session, + skip: int = 0, + limit: int = 100, + search: str | None = None, + is_active: bool | None = None, + is_verified: bool | None = None, + ) -> tuple[List[Company], int]: + """ + Get paginated list of companies with optional filters. + + Args: + db: Database session + skip: Number of records to skip + limit: Maximum number of records to return + search: Search term for company name + is_active: Filter by active status + is_verified: Filter by verified status + + Returns: + Tuple of (companies list, total count) + """ + query = select(Company).options(joinedload(Company.vendors)) + + # Apply filters + if search: + query = query.where(Company.name.ilike(f"%{search}%")) + if is_active is not None: + query = query.where(Company.is_active == is_active) + if is_verified is not None: + query = query.where(Company.is_verified == is_verified) + + # Get total count + count_query = select(func.count()).select_from(query.subquery()) + total = db.execute(count_query).scalar() + + # Apply pagination and order + query = query.order_by(Company.name).offset(skip).limit(limit) + + companies = list(db.execute(query).scalars().all()) + + return companies, total + + def update_company( + self, db: Session, company_id: int, company_data: CompanyUpdate + ) -> Company: + """ + Update company information. + + Args: + db: Database session + company_id: Company ID + company_data: Updated company data + + Returns: + Updated company + + Raises: + CompanyNotFoundException: If company not found + """ + company = self.get_company_by_id(db, company_id) + + # Update only provided fields + update_data = company_data.model_dump(exclude_unset=True) + for field, value in update_data.items(): + setattr(company, field, value) + + db.flush() + logger.info(f"Updated company ID {company_id}") + + return company + + def delete_company(self, db: Session, company_id: int) -> None: + """ + Delete a company and all associated vendors. + + Args: + db: Database session + company_id: Company ID + + Raises: + CompanyNotFoundException: If company not found + """ + company = self.get_company_by_id(db, company_id) + + # Due to cascade="all, delete-orphan", associated vendors will be deleted + db.delete(company) + db.flush() + logger.info(f"Deleted company ID {company_id} and associated vendors") + + def toggle_verification(self, db: Session, company_id: int, is_verified: bool) -> Company: + """ + Toggle company verification status. + + Args: + db: Database session + company_id: Company ID + is_verified: New verification status + + Returns: + Updated company + + Raises: + CompanyNotFoundException: If company not found + """ + company = self.get_company_by_id(db, company_id) + company.is_verified = is_verified + db.flush() + logger.info( + f"Company ID {company_id} verification set to {is_verified}" + ) + + return company + + def toggle_active(self, db: Session, company_id: int, is_active: bool) -> Company: + """ + Toggle company active status. + + Args: + db: Database session + company_id: Company ID + is_active: New active status + + Returns: + Updated company + + Raises: + CompanyNotFoundException: If company not found + """ + company = self.get_company_by_id(db, company_id) + company.is_active = is_active + db.flush() + logger.info( + f"Company ID {company_id} active status set to {is_active}" + ) + + return company + + def _generate_temp_password(self, length: int = 12) -> str: + """Generate secure temporary password.""" + alphabet = string.ascii_letters + string.digits + "!@#$%^&*" + return "".join(secrets.choice(alphabet) for _ in range(length)) + + +# Create service instance following the same pattern as other services +company_service = CompanyService() diff --git a/models/database/company.py b/models/database/company.py new file mode 100644 index 00000000..1ff986b1 --- /dev/null +++ b/models/database/company.py @@ -0,0 +1,106 @@ +# models/database/company.py +""" +Company model representing the business entity that owns one or more vendor brands. + +A Company represents the legal/business entity with contact information, +while Vendors represent the individual brands/storefronts operated by that company. +""" + +from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, Text +from sqlalchemy.orm import relationship + +from app.core.database import Base +from models.database.base import TimestampMixin + + +class Company(Base, TimestampMixin): + """ + Represents a company (business entity) in the system. + + A company owns one or more vendor brands. All business/contact information + is stored at the company level to avoid duplication. + """ + + __tablename__ = "companies" + + # ======================================================================== + # Basic Information + # ======================================================================== + id = Column(Integer, primary_key=True, index=True) + """Unique identifier for the company.""" + + name = Column(String, nullable=False, index=True) + """Company legal/business name.""" + + description = Column(Text) + """Optional description of the company.""" + + # ======================================================================== + # Ownership + # ======================================================================== + owner_user_id = Column(Integer, ForeignKey("users.id"), nullable=False) + """Foreign key to the user who owns this company.""" + + # ======================================================================== + # Contact Information + # ======================================================================== + contact_email = Column(String, nullable=False) + """Primary business contact email.""" + + contact_phone = Column(String) + """Business phone number.""" + + website = Column(String) + """Company website URL.""" + + # ======================================================================== + # Business Details + # ======================================================================== + business_address = Column(Text) + """Physical business address.""" + + tax_number = Column(String) + """Tax/VAT registration number.""" + + # ======================================================================== + # Status Flags + # ======================================================================== + is_active = Column(Boolean, default=True, nullable=False) + """Whether the company is active. Affects all associated vendors.""" + + is_verified = Column(Boolean, default=False, nullable=False) + """Whether the company has been verified by platform admins.""" + + # ======================================================================== + # Relationships + # ======================================================================== + owner = relationship("User", back_populates="owned_companies") + """The user who owns this company.""" + + vendors = relationship( + "Vendor", + back_populates="company", + cascade="all, delete-orphan", + order_by="Vendor.name", + ) + """All vendor brands operated by this company.""" + + def __repr__(self): + """String representation of the Company object.""" + return f"" + + # ======================================================================== + # Helper Properties + # ======================================================================== + + @property + def vendor_count(self) -> int: + """Get the number of vendors belonging to this company.""" + return len(self.vendors) if self.vendors else 0 + + @property + def active_vendor_count(self) -> int: + """Get the number of active vendors belonging to this company.""" + if not self.vendors: + return 0 + return sum(1 for v in self.vendors if v.is_active) diff --git a/models/database/vendor.py b/models/database/vendor.py index da36ca91..88a98cdc 100644 --- a/models/database/vendor.py +++ b/models/database/vendor.py @@ -35,48 +35,47 @@ class Vendor(Base, TimestampMixin): id = Column( Integer, primary_key=True, index=True ) # Primary key and indexed column for vendor ID + + # Company relationship + company_id = Column( + Integer, ForeignKey("companies.id"), nullable=False, index=True + ) # Foreign key to the parent company + vendor_code = Column( String, unique=True, index=True, nullable=False ) # Unique, indexed, non-nullable vendor code column subdomain = Column( String(100), unique=True, nullable=False, index=True ) # Unique, non-nullable subdomain column with indexing - name = Column(String, nullable=False) # Non-nullable name column for the vendor + name = Column(String, nullable=False) # Non-nullable name column for the vendor (brand name) description = Column(Text) # Optional text description column for the vendor + owner_user_id = Column( - Integer, ForeignKey("users.id"), nullable=False - ) # Foreign key to user ID of the vendor's owner + Integer, ForeignKey("users.id"), nullable=True + ) # Foreign key to user ID of the vendor's owner (DEPRECATED - use company.owner_user_id instead) - # Contact information - contact_email = Column(String) # Optional email column for contact information - contact_phone = Column(String) # Optional phone column for contact information - website = Column(String) # Optional website column for contact information - - # Letzshop URLs - multi-language support + # Letzshop URLs - multi-language support (brand-specific marketplace feeds) letzshop_csv_url_fr = Column(String) # URL for French CSV in Letzshop letzshop_csv_url_en = Column(String) # URL for English CSV in Letzshop letzshop_csv_url_de = Column(String) # URL for German CSV in Letzshop - # Business information - business_address = Column( - Text - ) # Optional text address column for business information - tax_number = Column(String) # Optional tax number column for business information - - # Status + # Status (vendor-specific, can differ from company status) is_active = Column( Boolean, default=True - ) # Boolean to indicate if the vendor is active + ) # Boolean to indicate if the vendor brand is active is_verified = Column( Boolean, default=False - ) # Boolean to indicate if the vendor is verified + ) # Boolean to indicate if the vendor brand is verified # ======================================================================== # Relationships # ======================================================================== + company = relationship( + "Company", back_populates="vendors" + ) # Relationship with Company model for the parent company owner = relationship( "User", back_populates="owned_vendors" - ) # Relationship with User model for the vendor's owner + ) # Relationship with User model for the vendor's owner (legacy) vendor_users = relationship( "VendorUser", back_populates="vendor" ) # Relationship with VendorUser model for users in this vendor diff --git a/models/schema/company.py b/models/schema/company.py new file mode 100644 index 00000000..b8a4a035 --- /dev/null +++ b/models/schema/company.py @@ -0,0 +1,155 @@ +# models/schema/company.py +""" +Pydantic schemas for Company model. + +These schemas are used for API request/response validation and serialization. +""" + +from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator + + +class CompanyBase(BaseModel): + """Base schema for company with common fields.""" + + name: str = Field(..., min_length=2, max_length=200, description="Company name") + description: str | None = Field(None, description="Company description") + contact_email: EmailStr = Field(..., description="Business contact email") + contact_phone: str | None = Field(None, description="Business phone number") + website: str | None = Field(None, description="Company website URL") + business_address: str | None = Field(None, description="Physical business address") + tax_number: str | None = Field(None, description="Tax/VAT registration number") + + @field_validator("contact_email") + @classmethod + def normalize_email(cls, v): + """Normalize email to lowercase.""" + return v.lower() if v else v + + +class CompanyCreate(CompanyBase): + """ + Schema for creating a new company. + + Requires owner_email to create the associated owner user account. + """ + + owner_email: EmailStr = Field( + ..., description="Email for the company owner account" + ) + + @field_validator("owner_email") + @classmethod + def normalize_owner_email(cls, v): + """Normalize owner email to lowercase.""" + return v.lower() if v else v + + model_config = ConfigDict(from_attributes=True) + + +class CompanyUpdate(BaseModel): + """ + Schema for updating company information. + + All fields are optional to support partial updates. + """ + + name: str | None = Field(None, min_length=2, max_length=200) + description: str | None = None + contact_email: EmailStr | None = None + contact_phone: str | None = None + website: str | None = None + business_address: str | None = None + tax_number: str | None = None + + # Status (Admin only) + is_active: bool | None = None + is_verified: bool | None = None + + @field_validator("contact_email") + @classmethod + def normalize_email(cls, v): + """Normalize email to lowercase.""" + return v.lower() if v else v + + model_config = ConfigDict(from_attributes=True) + + +class CompanyResponse(BaseModel): + """Standard schema for company response data.""" + + model_config = ConfigDict(from_attributes=True) + + id: int + name: str + description: str | None + + # Owner information + owner_user_id: int + + # Contact Information + contact_email: str + contact_phone: str | None + website: str | None + + # Business Information + business_address: str | None + tax_number: str | None + + # Status Flags + is_active: bool + is_verified: bool + + # Timestamps + created_at: str + updated_at: str + + +class CompanyDetailResponse(CompanyResponse): + """ + Detailed company response including vendor count. + + Used for company detail pages and admin views. + """ + + vendor_count: int = Field(0, description="Number of vendors under this company") + active_vendor_count: int = Field( + 0, description="Number of active vendors under this company" + ) + + +class CompanyListResponse(BaseModel): + """Schema for paginated company list.""" + + companies: list[CompanyResponse] + total: int + skip: int + limit: int + + +class CompanyCreateResponse(BaseModel): + """ + Response after creating a company with owner account. + + Includes temporary password for the owner (shown only once). + """ + + company: CompanyResponse + owner_user_id: int + owner_username: str + owner_email: str + temporary_password: str = Field( + ..., description="Temporary password for owner (SHOWN ONLY ONCE)" + ) + login_url: str | None = Field(None, description="URL for company owner to login") + + +class CompanySummary(BaseModel): + """Lightweight company summary for dropdowns and quick references.""" + + model_config = ConfigDict(from_attributes=True) + + id: int + name: str + is_active: bool + is_verified: bool + vendor_count: int = 0