fix: move database queries from API to service layer

Move direct database queries in subscriptions.py to admin_subscription_service:
- Add get_vendor() method to verify vendor exists
- Add get_vendor_usage_counts() method for products/team counts
- Update API endpoints to use service methods
- Remove unused imports (func, Product, VendorUser)

Fixes architecture validation errors [API-002]

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2025-12-26 21:01:01 +01:00
parent 25279a03d4
commit a6f9aa9742
2 changed files with 46 additions and 65 deletions

View File

@@ -21,13 +21,14 @@ from app.exceptions import (
ResourceNotFoundException,
TierNotFoundException,
)
from models.database.product import Product
from models.database.subscription import (
BillingHistory,
SubscriptionStatus,
SubscriptionTier,
VendorSubscription,
)
from models.database.vendor import Vendor
from models.database.vendor import Vendor, VendorUser
logger = logging.getLogger(__name__)
@@ -197,6 +198,39 @@ class AdminSubscriptionService:
return sub, vendor
def get_vendor(self, db: Session, vendor_id: int) -> Vendor:
"""Get a vendor by ID."""
vendor = db.query(Vendor).filter(Vendor.id == vendor_id).first()
if not vendor:
raise ResourceNotFoundException("Vendor", str(vendor_id))
return vendor
def get_vendor_usage_counts(self, db: Session, vendor_id: int) -> dict:
"""Get usage counts (products and team members) for a vendor."""
products_count = (
db.query(func.count(Product.id))
.filter(Product.vendor_id == vendor_id)
.scalar()
or 0
)
team_count = (
db.query(func.count(VendorUser.id))
.filter(
VendorUser.vendor_id == vendor_id,
VendorUser.is_active == True, # noqa: E712
)
.scalar()
or 0
)
return {
"products_count": products_count,
"team_count": team_count,
}
# =========================================================================
# Billing History
# =========================================================================