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
This commit is contained in:
@@ -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)
|
||||
297
app/api/v1/admin/companies.py
Normal file
297
app/api/v1/admin/companies.py
Normal file
@@ -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"}
|
||||
128
app/exceptions/company.py
Normal file
128
app/exceptions/company.py
Normal file
@@ -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},
|
||||
)
|
||||
266
app/services/company_service.py
Normal file
266
app/services/company_service.py
Normal file
@@ -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()
|
||||
106
models/database/company.py
Normal file
106
models/database/company.py
Normal file
@@ -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"<Company(id={self.id}, name='{self.name}', vendors={len(self.vendors) if self.vendors else 0})>"
|
||||
|
||||
# ========================================================================
|
||||
# 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)
|
||||
@@ -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
|
||||
|
||||
155
models/schema/company.py
Normal file
155
models/schema/company.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user