refactor: update services for company-centric ownership

- Update admin_service: remove vendor owner methods, fix get_all_vendors query
- Update company_service: add transfer_ownership method
- Update vendor_service: check ownership via company relationship
- Remove backwards compatibility code for Vendor.owner_user_id

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
2025-12-02 19:39:23 +01:00
parent df8731d7ae
commit ec2885dab5
3 changed files with 187 additions and 292 deletions

View File

@@ -13,10 +13,10 @@ from typing import List, Optional
from sqlalchemy import func, select
from sqlalchemy.orm import Session, joinedload
from app.exceptions import CompanyNotFoundException
from app.exceptions import CompanyNotFoundException, UserNotFoundException
from models.database.company import Company
from models.database.user import User
from models.schema.company import CompanyCreate, CompanyUpdate
from models.schema.company import CompanyCreate, CompanyTransferOwnership, CompanyUpdate
logger = logging.getLogger(__name__)
@@ -110,7 +110,7 @@ class CompanyService:
select(Company)
.where(Company.id == company_id)
.options(joinedload(Company.vendors))
).scalar_one_or_none()
).unique().scalar_one_or_none()
if not company:
raise CompanyNotFoundException(company_id)
@@ -257,6 +257,69 @@ class CompanyService:
return company
def transfer_ownership(
self,
db: Session,
company_id: int,
transfer_data: CompanyTransferOwnership,
) -> tuple[Company, User, User]:
"""
Transfer company ownership to another user.
This is a critical operation that:
- Changes the company's owner_user_id
- All vendors under the company automatically inherit the new owner
- Logs the transfer for audit purposes
Args:
db: Database session
company_id: Company ID
transfer_data: Transfer ownership data
Returns:
Tuple of (company, old_owner, new_owner)
Raises:
CompanyNotFoundException: If company not found
UserNotFoundException: If new owner user not found
ValueError: If trying to transfer to current owner
"""
# Get company
company = self.get_company_by_id(db, company_id)
old_owner_id = company.owner_user_id
# Get old owner
old_owner = db.execute(
select(User).where(User.id == old_owner_id)
).scalar_one_or_none()
if not old_owner:
raise UserNotFoundException(str(old_owner_id))
# Get new owner
new_owner = db.execute(
select(User).where(User.id == transfer_data.new_owner_user_id)
).scalar_one_or_none()
if not new_owner:
raise UserNotFoundException(str(transfer_data.new_owner_user_id))
# Prevent transferring to same owner
if old_owner_id == transfer_data.new_owner_user_id:
raise ValueError("Cannot transfer ownership to the current owner")
# Update company owner (vendors inherit ownership via company relationship)
company.owner_user_id = new_owner.id
db.flush()
logger.info(
f"Company {company.id} ({company.name}) ownership transferred "
f"from user {old_owner.id} ({old_owner.email}) "
f"to user {new_owner.id} ({new_owner.email}). "
f"Reason: {transfer_data.transfer_reason or 'Not specified'}"
)
return company, old_owner, new_owner
def _generate_temp_password(self, length: int = 12) -> str:
"""Generate secure temporary password."""
alphabet = string.ascii_letters + string.digits + "!@#$%^&*"