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

style: apply black and isort formatting across entire codebase

Browse Source 
- Standardize quote style (single to double quotes)
- Reorder and group imports alphabetically
- Fix line breaks and indentation for consistency
- Apply PEP 8 formatting standards

Also updated Makefile to exclude both venv and .venv from code quality checks.

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

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2025-11-28 19:30:17 +01:00
parent 13f0094743
commit 21c13ca39b
236 changed files with 8450 additions and 6545 deletions
Download Patch File Download Diff File Expand all files Collapse all files

168
app/services/vendor_domain_service.py
View File

@@ -12,30 +12,28 @@ This module provides classes and functions for:
import logging
import secrets
from typing import List, Tuple, Optional
from datetime import datetime, timezone
from typing import List, Optional, Tuple
from sqlalchemy.orm import Session
from sqlalchemy import and_
from sqlalchemy.orm import Session
from app.exceptions import (
VendorNotFoundException,
VendorDomainNotFoundException,
VendorDomainAlreadyExistsException,
InvalidDomainFormatException,
ReservedDomainException,
DomainNotVerifiedException,
DomainVerificationFailedException,
DomainAlreadyVerifiedException,
MultiplePrimaryDomainsException,
DNSVerificationException,
MaxDomainsReachedException,
UnauthorizedDomainAccessException,
ValidationException,
)
from models.schema.vendor_domain import VendorDomainCreate, VendorDomainUpdate
from app.exceptions import (DNSVerificationException,
DomainAlreadyVerifiedException,
DomainNotVerifiedException,
DomainVerificationFailedException,
InvalidDomainFormatException,
MaxDomainsReachedException,
MultiplePrimaryDomainsException,
ReservedDomainException,
UnauthorizedDomainAccessException,
ValidationException,
VendorDomainAlreadyExistsException,
VendorDomainNotFoundException,
VendorNotFoundException)
from models.database.vendor import Vendor
from models.database.vendor_domain import VendorDomain
from models.schema.vendor_domain import VendorDomainCreate, VendorDomainUpdate
logger = logging.getLogger(__name__)
@@ -45,13 +43,19 @@ class VendorDomainService:
def __init__(self):
self.max_domains_per_vendor = 10 # Configure as needed
self.reserved_subdomains = ['www', 'admin', 'api', 'mail', 'smtp', 'ftp', 'cpanel', 'webmail']
self.reserved_subdomains = [
"www",
"admin",
"api",
"mail",
"smtp",
"ftp",
"cpanel",
"webmail",
]
def add_domain(
self,
db: Session,
vendor_id: int,
domain_data: VendorDomainCreate
self, db: Session, vendor_id: int, domain_data: VendorDomainCreate
) -> VendorDomain:
"""
Add a custom domain to vendor.
@@ -85,12 +89,14 @@ class VendorDomainService:
# Check if domain already exists
if self._domain_exists(db, normalized_domain):
existing_domain = db.query(VendorDomain).filter(
VendorDomain.domain == normalized_domain
).first()
existing_domain = (
db.query(VendorDomain)
.filter(VendorDomain.domain == normalized_domain)
.first()
)
raise VendorDomainAlreadyExistsException(
normalized_domain,
existing_domain.vendor_id if existing_domain else None
existing_domain.vendor_id if existing_domain else None,
)
# If setting as primary, unset other primary domains
@@ -104,8 +110,8 @@ class VendorDomainService:
is_primary=domain_data.is_primary,
verification_token=secrets.token_urlsafe(32),
is_verified=False, # Requires DNS verification
is_active=False, # Cannot be active until verified
ssl_status="pending"
is_active=False, # Cannot be active until verified
ssl_status="pending",
)
db.add(new_domain)
@@ -120,7 +126,7 @@ class VendorDomainService:
VendorDomainAlreadyExistsException,
MaxDomainsReachedException,
InvalidDomainFormatException,
ReservedDomainException
ReservedDomainException,
):
db.rollback()
raise
@@ -129,11 +135,7 @@ class VendorDomainService:
logger.error(f"Error adding domain: {str(e)}")
raise ValidationException("Failed to add domain")
def get_vendor_domains(
self,
db: Session,
vendor_id: int
) -> List[VendorDomain]:
def get_vendor_domains(self, db: Session, vendor_id: int) -> List[VendorDomain]:
"""
Get all domains for a vendor.
@@ -151,12 +153,14 @@ class VendorDomainService:
# Verify vendor exists
self._get_vendor_by_id_or_raise(db, vendor_id)
domains = db.query(VendorDomain).filter(
VendorDomain.vendor_id == vendor_id
).order_by(
VendorDomain.is_primary.desc(),
VendorDomain.created_at.desc()
).all()
domains = (
db.query(VendorDomain)
.filter(VendorDomain.vendor_id == vendor_id)
.order_by(
VendorDomain.is_primary.desc(), VendorDomain.created_at.desc()
)
.all()
)
return domains
@@ -166,11 +170,7 @@ class VendorDomainService:
logger.error(f"Error getting vendor domains: {str(e)}")
raise ValidationException("Failed to retrieve domains")
def get_domain_by_id(
self,
db: Session,
domain_id: int
) -> VendorDomain:
def get_domain_by_id(self, db: Session, domain_id: int) -> VendorDomain:
"""
Get domain by ID.
@@ -190,10 +190,7 @@ class VendorDomainService:
return domain
def update_domain(
self,
db: Session,
domain_id: int,
domain_update: VendorDomainUpdate
self, db: Session, domain_id: int, domain_update: VendorDomainUpdate
) -> VendorDomain:
"""
Update domain settings.
@@ -215,7 +212,9 @@ class VendorDomainService:
# If setting as primary, unset other primary domains
if domain_update.is_primary:
self._unset_primary_domains(db, domain.vendor_id, exclude_domain_id=domain_id)
self._unset_primary_domains(
db, domain.vendor_id, exclude_domain_id=domain_id
)
domain.is_primary = True
# If activating, check verification
@@ -240,11 +239,7 @@ class VendorDomainService:
logger.error(f"Error updating domain: {str(e)}")
raise ValidationException("Failed to update domain")
def delete_domain(
self,
db: Session,
domain_id: int
) -> str:
def delete_domain(self, db: Session, domain_id: int) -> str:
"""
Delete a custom domain.
@@ -277,11 +272,7 @@ class VendorDomainService:
logger.error(f"Error deleting domain: {str(e)}")
raise ValidationException("Failed to delete domain")
def verify_domain(
self,
db: Session,
domain_id: int
) -> Tuple[VendorDomain, str]:
def verify_domain(self, db: Session, domain_id: int) -> Tuple[VendorDomain, str]:
"""
Verify domain ownership via DNS TXT record.
@@ -313,8 +304,7 @@ class VendorDomainService:
# Query DNS TXT records
try:
txt_records = dns.resolver.resolve(
f"_wizamart-verify.{domain.domain}",
'TXT'
f"_wizamart-verify.{domain.domain}", "TXT"
)
# Check if verification token is present
@@ -332,42 +322,33 @@ class VendorDomainService:
# Token not found
raise DomainVerificationFailedException(
domain.domain,
"Verification token not found in DNS records"
domain.domain, "Verification token not found in DNS records"
)
except dns.resolver.NXDOMAIN:
raise DomainVerificationFailedException(
domain.domain,
f"DNS record _wizamart-verify.{domain.domain} not found"
f"DNS record _wizamart-verify.{domain.domain} not found",
)
except dns.resolver.NoAnswer:
raise DomainVerificationFailedException(
domain.domain,
"No TXT records found for verification"
domain.domain, "No TXT records found for verification"
)
except Exception as dns_error:
raise DNSVerificationException(
domain.domain,
str(dns_error)
)
raise DNSVerificationException(domain.domain, str(dns_error))
except (
VendorDomainNotFoundException,
DomainAlreadyVerifiedException,
DomainVerificationFailedException,
DNSVerificationException
DNSVerificationException,
):
raise
except Exception as e:
logger.error(f"Error verifying domain: {str(e)}")
raise ValidationException("Failed to verify domain")
def get_verification_instructions(
self,
db: Session,
domain_id: int
) -> dict:
def get_verification_instructions(self, db: Session, domain_id: int) -> dict:
"""
Get DNS verification instructions for domain.
@@ -390,20 +371,20 @@ class VendorDomainService:
"step1": "Go to your domain's DNS settings (at your domain registrar)",
"step2": "Add a new TXT record with the following values:",
"step3": "Wait for DNS propagation (5-15 minutes)",
"step4": "Click 'Verify Domain' button in admin panel"
"step4": "Click 'Verify Domain' button in admin panel",
},
"txt_record": {
"type": "TXT",
"name": "_wizamart-verify",
"value": domain.verification_token,
"ttl": 3600
"ttl": 3600,
},
"common_registrars": {
"Cloudflare": "https://dash.cloudflare.com",
"GoDaddy": "https://dcc.godaddy.com/manage/dns",
"Namecheap": "https://www.namecheap.com/myaccount/domain-list/",
"Google Domains": "https://domains.google.com"
}
"Google Domains": "https://domains.google.com",
},
}
# Private helper methods
@@ -416,36 +397,33 @@ class VendorDomainService:
def _check_domain_limit(self, db: Session, vendor_id: int) -> None:
"""Check if vendor has reached maximum domain limit."""
domain_count = db.query(VendorDomain).filter(
VendorDomain.vendor_id == vendor_id
).count()
domain_count = (
db.query(VendorDomain).filter(VendorDomain.vendor_id == vendor_id).count()
)
if domain_count >= self.max_domains_per_vendor:
raise MaxDomainsReachedException(vendor_id, self.max_domains_per_vendor)
def _domain_exists(self, db: Session, domain: str) -> bool:
"""Check if domain already exists in system."""
return db.query(VendorDomain).filter(
VendorDomain.domain == domain
).first() is not None
return (
db.query(VendorDomain).filter(VendorDomain.domain == domain).first()
is not None
)
def _validate_domain_format(self, domain: str) -> None:
"""Validate domain format and check for reserved subdomains."""
# Check for reserved subdomains
first_part = domain.split('.')[0]
first_part = domain.split(".")[0]
if first_part in self.reserved_subdomains:
raise ReservedDomainException(domain, first_part)
def _unset_primary_domains(
self,
db: Session,
vendor_id: int,
exclude_domain_id: Optional[int] = None
self, db: Session, vendor_id: int, exclude_domain_id: Optional[int] = None
) -> None:
"""Unset all primary domains for vendor."""
query = db.query(VendorDomain).filter(
VendorDomain.vendor_id == vendor_id,
VendorDomain.is_primary == True
VendorDomain.vendor_id == vendor_id, VendorDomain.is_primary == True
)
if exclude_domain_id: