sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: update API endpoints for company-vendor architecture

Browse Source 
- Add transfer-ownership endpoint to companies API
- Add user search endpoint for autocomplete (/admin/users/search)
- Update company detail endpoint to include owner info and vendors list
- Update vendor endpoints to use company relationship for ownership
- Update deps.py vendor access check to use company.owner_user_id

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

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2025-12-02 19:39:31 +01:00
parent ec2885dab5
commit 66c967a04e
6 changed files with 194 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
app/api/deps.py
View File

@@ -513,13 +513,20 @@ def get_user_vendor(
VendorNotFoundException: If vendor doesn't exist
UnauthorizedVendorAccessException: If user doesn't have access
"""
vendor = db.query(Vendor).filter(Vendor.vendor_code == vendor_code.upper()).first()
from sqlalchemy.orm import joinedload
vendor = (
db.query(Vendor)
.options(joinedload(Vendor.company))
.filter(Vendor.vendor_code == vendor_code.upper())
.first()
)
if not vendor:
raise VendorNotFoundException(vendor_code)
# Check if user owns this vendor
if vendor.owner_user_id == current_user.id:
# Check if user owns this vendor (via company ownership)
if vendor.company and vendor.company.owner_user_id == current_user.id:
return vendor
# Check if user is team member

68
app/api/v1/admin/companies.py
View File

@@ -4,6 +4,7 @@ Company management endpoints for admin.
"""
import logging
from datetime import UTC, datetime
from fastapi import APIRouter, Body, Depends, Path, Query
from sqlalchemy.orm import Session
@@ -19,6 +20,8 @@ from models.schema.company import (
CompanyDetailResponse,
CompanyListResponse,
CompanyResponse,
CompanyTransferOwnership,
CompanyTransferOwnershipResponse,
CompanyUpdate,
)
@@ -136,11 +139,26 @@ def get_company_details(
vendor_count = len(company.vendors)
active_vendor_count = sum(1 for v in company.vendors if v.is_active)
# Build vendors list for detail view
vendors_list = [
{
"id": v.id,
"vendor_code": v.vendor_code,
"name": v.name,
"subdomain": v.subdomain,
"is_active": v.is_active,
"is_verified": v.is_verified,
}
for v in company.vendors
]
return CompanyDetailResponse(
id=company.id,
name=company.name,
description=company.description,
owner_user_id=company.owner_user_id,
owner_email=company.owner.email if company.owner else None,
owner_username=company.owner.username if company.owner else None,
contact_email=company.contact_email,
contact_phone=company.contact_phone,
website=company.website,
@@ -152,6 +170,7 @@ def get_company_details(
updated_at=company.updated_at.isoformat(),
vendor_count=vendor_count,
active_vendor_count=active_vendor_count,
vendors=vendors_list,
)
@@ -260,6 +279,55 @@ def toggle_company_status(
)
@router.post(
"/{company_id}/transfer-ownership",
response_model=CompanyTransferOwnershipResponse,
)
def transfer_company_ownership(
company_id: int = Path(..., description="Company ID"),
transfer_data: CompanyTransferOwnership = Body(...),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
):
"""
Transfer company ownership to another user (Admin only).
**This is a critical operation that:**
- Changes the company's owner_user_id
- Updates all associated vendors' owner_user_id
- Creates audit trail
⚠️ **This action is logged and should be used carefully.**
**Requires:**
- `new_owner_user_id`: ID of user who will become owner
- `confirm_transfer`: Must be true
- `transfer_reason`: Optional reason for audit trail
"""
company, old_owner, new_owner = company_service.transfer_ownership(
db, company_id, transfer_data
)
db.commit() # ✅ ARCH: Commit at API level for transaction control
return CompanyTransferOwnershipResponse(
message="Ownership transferred successfully",
company_id=company.id,
company_name=company.name,
old_owner={
"id": old_owner.id,
"username": old_owner.username,
"email": old_owner.email,
},
new_owner={
"id": new_owner.id,
"username": new_owner.username,
"email": new_owner.email,
},
transferred_at=datetime.now(UTC),
transfer_reason=transfer_data.transfer_reason,
)
@router.delete("/{company_id}")
def delete_company(
company_id: int = Path(..., description="Company ID"),

35
app/api/v1/admin/users.py
View File

@@ -49,3 +49,38 @@ def get_user_statistics(
):
"""Get user statistics for admin dashboard (Admin only)."""
return stats_service.get_user_statistics(db)
@router.get("/search")
def search_users(
q: str = Query(..., min_length=2, description="Search query (username or email)"),
limit: int = Query(10, ge=1, le=50),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
):
"""
Search users by username or email (Admin only).
Used for autocomplete in ownership transfer.
"""
search_term = f"%{q.lower()}%"
users = (
db.query(User)
.filter(
(User.username.ilike(search_term)) | (User.email.ilike(search_term))
)
.limit(limit)
.all()
)
return {
"users": [
{
"id": user.id,
"username": user.username,
"email": user.email,
"is_active": user.is_active,
}
for user in users
]
}

184
app/api/v1/admin/vendors.py
View File

@@ -4,17 +4,17 @@ Vendor management endpoints for admin.
"""
import logging
from datetime import UTC
from fastapi import APIRouter, Body, Depends, Path, Query
from sqlalchemy import func
from sqlalchemy.orm import Session
from sqlalchemy.orm import Session, joinedload
from app.api.deps import get_current_admin_api
from app.core.database import get_db
from app.exceptions import ConfirmationRequiredException, VendorNotFoundException
from app.services.admin_service import admin_service
from app.services.stats_service import stats_service
from models.database.company import Company
from models.database.user import User
from models.database.vendor import Vendor
from models.schema.stats import VendorStatsResponse
@@ -23,8 +23,6 @@ from models.schema.vendor import (
VendorCreateResponse,
VendorDetailResponse,
VendorListResponse,
VendorTransferOwnership,
VendorTransferOwnershipResponse,
VendorUpdate,
)
@@ -61,6 +59,7 @@ def _get_vendor_by_identifier(db: Session, identifier: str) -> Vendor:
# Try as vendor_code (case-insensitive)
vendor = (
db.query(Vendor)
.options(joinedload(Vendor.company).joinedload(Company.owner))
.filter(func.upper(Vendor.vendor_code) == identifier.upper())
.first()
)
@@ -72,30 +71,22 @@ def _get_vendor_by_identifier(db: Session, identifier: str) -> Vendor:
@router.post("", response_model=VendorCreateResponse)
def create_vendor_with_owner(
def create_vendor(
vendor_data: VendorCreate,
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
):
"""
Create a new vendor with owner user account (Admin only).
Create a new vendor (storefront/brand) under an existing company (Admin only).
This endpoint:
1. Creates a new vendor record
2. Creates an owner user account with owner_email
3. Sets contact_email (defaults to owner_email if not provided)
4. Sets up default roles (Owner, Manager, Editor, Viewer)
5. Returns credentials (temporary password shown ONCE)
1. Validates that the parent company exists
2. Creates a new vendor record linked to the company
3. Sets up default roles (Owner, Manager, Editor, Viewer)
**Email Fields:**
- `owner_email`: Used for owner's login/authentication (stored in users.email)
- `contact_email`: Public business contact (stored in vendors.contact_email)
Returns vendor details with owner credentials.
The vendor inherits owner and contact information from its parent company.
"""
vendor, owner_user, temp_password = admin_service.create_vendor_with_owner(
db=db, vendor_data=vendor_data
)
vendor = admin_service.create_vendor(db=db, vendor_data=vendor_data)
return VendorCreateResponse(
# Vendor fields
@@ -104,12 +95,7 @@ def create_vendor_with_owner(
subdomain=vendor.subdomain,
name=vendor.name,
description=vendor.description,
owner_user_id=vendor.owner_user_id,
contact_email=vendor.contact_email,
contact_phone=vendor.contact_phone,
website=vendor.website,
business_address=vendor.business_address,
tax_number=vendor.tax_number,
company_id=vendor.company_id,
letzshop_csv_url_fr=vendor.letzshop_csv_url_fr,
letzshop_csv_url_en=vendor.letzshop_csv_url_en,
letzshop_csv_url_de=vendor.letzshop_csv_url_de,
@@ -117,10 +103,14 @@ def create_vendor_with_owner(
is_verified=vendor.is_verified,
created_at=vendor.created_at,
updated_at=vendor.updated_at,
# Owner credentials
owner_email=owner_user.email,
owner_username=owner_user.username,
temporary_password=temp_password,
# Company info
company_name=vendor.company.name,
company_contact_email=vendor.company.contact_email,
company_contact_phone=vendor.company.contact_phone,
company_website=vendor.company.website,
# Owner info (from company)
owner_email=vendor.company.owner.email,
owner_username=vendor.company.owner.username,
login_url=f"http://localhost:8000/vendor/{vendor.subdomain}/login",
)
@@ -170,13 +160,11 @@ def get_vendor_details(
current_admin: User = Depends(get_current_admin_api),
):
"""
Get detailed vendor information including owner details (Admin only).
Get detailed vendor information including company and owner details (Admin only).
Accepts either vendor ID (integer) or vendor_code (string).
Returns both:
- `contact_email` (business contact)
- `owner_email` (owner's authentication email)
Returns vendor info with company contact details and owner info.
"""
vendor = _get_vendor_by_identifier(db, vendor_identifier)
@@ -187,12 +175,7 @@ def get_vendor_details(
subdomain=vendor.subdomain,
name=vendor.name,
description=vendor.description,
owner_user_id=vendor.owner_user_id,
contact_email=vendor.contact_email,
contact_phone=vendor.contact_phone,
website=vendor.website,
business_address=vendor.business_address,
tax_number=vendor.tax_number,
company_id=vendor.company_id,
letzshop_csv_url_fr=vendor.letzshop_csv_url_fr,
letzshop_csv_url_en=vendor.letzshop_csv_url_en,
letzshop_csv_url_de=vendor.letzshop_csv_url_de,
@@ -200,9 +183,14 @@ def get_vendor_details(
is_verified=vendor.is_verified,
created_at=vendor.created_at,
updated_at=vendor.updated_at,
# Owner details
owner_email=vendor.owner.email,
owner_username=vendor.owner.username,
# Company info
company_name=vendor.company.name,
company_contact_email=vendor.company.contact_email,
company_contact_phone=vendor.company.contact_phone,
company_website=vendor.company.website,
# Owner details (from company)
owner_email=vendor.company.owner.email,
owner_username=vendor.company.owner.username,
)
@@ -220,15 +208,12 @@ def update_vendor(
**Can update:**
- Basic info: name, description, subdomain
- Business contact: contact_email, contact_phone, website
- Business details: business_address, tax_number
- Marketplace URLs
- Status: is_active, is_verified
**Cannot update:**
- `owner_email` (use POST /vendors/{id}/transfer-ownership)
- `vendor_code` (immutable)
- `owner_user_id` (use POST /vendors/{id}/transfer-ownership)
- Business contact info (use company update endpoints)
"""
vendor = _get_vendor_by_identifier(db, vendor_identifier)
vendor = admin_service.update_vendor(db, vendor.id, vendor_update)
@@ -239,12 +224,7 @@ def update_vendor(
subdomain=vendor.subdomain,
name=vendor.name,
description=vendor.description,
owner_user_id=vendor.owner_user_id,
contact_email=vendor.contact_email,
contact_phone=vendor.contact_phone,
website=vendor.website,
business_address=vendor.business_address,
tax_number=vendor.tax_number,
company_id=vendor.company_id,
letzshop_csv_url_fr=vendor.letzshop_csv_url_fr,
letzshop_csv_url_en=vendor.letzshop_csv_url_en,
letzshop_csv_url_de=vendor.letzshop_csv_url_de,
@@ -252,64 +232,20 @@ def update_vendor(
is_verified=vendor.is_verified,
created_at=vendor.created_at,
updated_at=vendor.updated_at,
owner_email=vendor.owner.email,
owner_username=vendor.owner.username,
# Company info
company_name=vendor.company.name,
company_contact_email=vendor.company.contact_email,
company_contact_phone=vendor.company.contact_phone,
company_website=vendor.company.website,
# Owner details (from company)
owner_email=vendor.company.owner.email,
owner_username=vendor.company.owner.username,
)
@router.post(
"/{vendor_identifier}/transfer-ownership",
response_model=VendorTransferOwnershipResponse,
)
def transfer_vendor_ownership(
vendor_identifier: str = Path(..., description="Vendor ID or vendor_code"),
transfer_data: VendorTransferOwnership = Body(...),
db: Session = Depends(get_db),
current_admin: User = Depends(get_current_admin_api),
):
"""
Transfer vendor ownership to another user (Admin only).
Accepts either vendor ID (integer) or vendor_code (string).
**This is a critical operation that:**
- Changes the owner_user_id
- Assigns new owner to "Owner" role
- Demotes old owner to "Manager" role (or removes them)
- Creates audit trail
⚠️ **This action is logged and should be used carefully.**
**Requires:**
- `new_owner_user_id`: ID of user who will become owner
- `confirm_transfer`: Must be true
- `transfer_reason`: Optional reason for audit trail
"""
from datetime import datetime
vendor = _get_vendor_by_identifier(db, vendor_identifier)
vendor, old_owner, new_owner = admin_service.transfer_vendor_ownership(
db, vendor.id, transfer_data
)
return VendorTransferOwnershipResponse(
message="Ownership transferred successfully",
vendor_id=vendor.id,
vendor_code=vendor.vendor_code,
vendor_name=vendor.name,
old_owner={
"id": old_owner.id,
"username": old_owner.username,
"email": old_owner.email,
},
new_owner={
"id": new_owner.id,
"username": new_owner.username,
"email": new_owner.email,
},
transferred_at=datetime.now(UTC),
transfer_reason=transfer_data.transfer_reason,
)
# NOTE: Ownership transfer is now at the Company level.
# Use PUT /api/v1/admin/companies/{id}/transfer-ownership instead.
# This endpoint is kept for backwards compatibility but may be removed in future versions.
@router.put("/{vendor_identifier}/verification", response_model=VendorDetailResponse)
@@ -345,12 +281,7 @@ def toggle_vendor_verification(
subdomain=vendor.subdomain,
name=vendor.name,
description=vendor.description,
owner_user_id=vendor.owner_user_id,
contact_email=vendor.contact_email,
contact_phone=vendor.contact_phone,
website=vendor.website,
business_address=vendor.business_address,
tax_number=vendor.tax_number,
company_id=vendor.company_id,
letzshop_csv_url_fr=vendor.letzshop_csv_url_fr,
letzshop_csv_url_en=vendor.letzshop_csv_url_en,
letzshop_csv_url_de=vendor.letzshop_csv_url_de,
@@ -358,8 +289,14 @@ def toggle_vendor_verification(
is_verified=vendor.is_verified,
created_at=vendor.created_at,
updated_at=vendor.updated_at,
owner_email=vendor.owner.email,
owner_username=vendor.owner.username,
# Company info
company_name=vendor.company.name,
company_contact_email=vendor.company.contact_email,
company_contact_phone=vendor.company.contact_phone,
company_website=vendor.company.website,
# Owner details (from company)
owner_email=vendor.company.owner.email,
owner_username=vendor.company.owner.username,
)
@@ -396,12 +333,7 @@ def toggle_vendor_status(
subdomain=vendor.subdomain,
name=vendor.name,
description=vendor.description,
owner_user_id=vendor.owner_user_id,
contact_email=vendor.contact_email,
contact_phone=vendor.contact_phone,
website=vendor.website,
business_address=vendor.business_address,
tax_number=vendor.tax_number,
company_id=vendor.company_id,
letzshop_csv_url_fr=vendor.letzshop_csv_url_fr,
letzshop_csv_url_en=vendor.letzshop_csv_url_en,
letzshop_csv_url_de=vendor.letzshop_csv_url_de,
@@ -409,8 +341,14 @@ def toggle_vendor_status(
is_verified=vendor.is_verified,
created_at=vendor.created_at,
updated_at=vendor.updated_at,
owner_email=vendor.owner.email,
owner_username=vendor.owner.username,
# Company info
company_name=vendor.company.name,
company_contact_email=vendor.company.contact_email,
company_contact_phone=vendor.company.contact_phone,
company_website=vendor.company.website,
# Owner details (from company)
owner_email=vendor.company.owner.email,
owner_username=vendor.company.owner.username,
)

19
app/api/v1/vendor/auth.py vendored
View File

@@ -91,8 +91,8 @@ def vendor_login(
vendor_role = "Member"
if vendor:
# Check if user is vendor owner
is_owner = vendor.owner_user_id == user.id
# Check if user is vendor owner (via company ownership)
is_owner = vendor.company and vendor.company.owner_user_id == user.id
if is_owner:
vendor_role = "Owner"
@@ -121,12 +121,15 @@ def vendor_login(
)
else:
# No vendor context - find which vendor this user belongs to
# Check owned vendors first
if user.owned_vendors:
vendor = user.owned_vendors[0]
vendor_role = "Owner"
# Check vendor memberships
elif user.vendor_memberships:
# Check owned vendors first (via company ownership)
for company in user.owned_companies:
if company.vendors:
vendor = company.vendors[0]
vendor_role = "Owner"
break
# Check vendor memberships if no owned vendor found
if not vendor and user.vendor_memberships:
active_membership = next(
(vm for vm in user.vendor_memberships if vm.is_active), None
)

18
app/api/v1/vendor/info.py vendored
View File

@@ -92,12 +92,7 @@ def get_vendor_info(
subdomain=vendor.subdomain,
name=vendor.name,
description=vendor.description,
owner_user_id=vendor.owner_user_id,
contact_email=vendor.contact_email,
contact_phone=vendor.contact_phone,
website=vendor.website,
business_address=vendor.business_address,
tax_number=vendor.tax_number,
company_id=vendor.company_id,
letzshop_csv_url_fr=vendor.letzshop_csv_url_fr,
letzshop_csv_url_en=vendor.letzshop_csv_url_en,
letzshop_csv_url_de=vendor.letzshop_csv_url_de,
@@ -105,7 +100,12 @@ def get_vendor_info(
is_verified=vendor.is_verified,
created_at=vendor.created_at,
updated_at=vendor.updated_at,
# Owner details
owner_email=vendor.owner.email,
owner_username=vendor.owner.username,
# Company info
company_name=vendor.company.name,
company_contact_email=vendor.company.contact_email,
company_contact_phone=vendor.company.contact_phone,
company_website=vendor.company.website,
# Owner details (from company)
owner_email=vendor.company.owner.email,
owner_username=vendor.company.owner.username,
)