Samir Boulahtit
238c1ec9b8
- Replace black, isort, and flake8 with Ruff (all-in-one linter and formatter) - Add comprehensive pyproject.toml configuration - Simplify Makefile code quality targets - Configure exclusions for venv/.venv in pyproject.toml - Auto-fix 1,359 linting issues across codebase Benefits: - Much faster builds (Ruff is written in Rust) - Single tool replaces multiple tools - More comprehensive rule set (UP, B, C4, SIM, PIE, RET, Q) - All configuration centralized in pyproject.toml - Better import sorting and formatting consistency 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
361 lines
13 KiB
Python
361 lines
13 KiB
Python
# models/database/vendor.py
|
|
"""
|
|
Vendor model representing entities that sell products or services.
|
|
|
|
This module defines the Vendor model along with its relationships to
|
|
other models such as User (owner), Product, Customer, Order, and MarketplaceImportJob.
|
|
"""
|
|
|
|
import enum
|
|
|
|
from sqlalchemy import (
|
|
JSON,
|
|
Boolean,
|
|
Column,
|
|
DateTime,
|
|
ForeignKey,
|
|
Integer,
|
|
String,
|
|
Text,
|
|
)
|
|
from sqlalchemy.orm import relationship
|
|
|
|
from app.core.config import settings
|
|
|
|
# Import Base from the central database module instead of creating a new one
|
|
from app.core.database import Base
|
|
from models.database.base import TimestampMixin
|
|
|
|
|
|
class Vendor(Base, TimestampMixin):
|
|
"""Represents a vendor in the system."""
|
|
|
|
__tablename__ = "vendors" # Name of the table in the database
|
|
|
|
id = Column(
|
|
Integer, primary_key=True, index=True
|
|
) # Primary key and indexed column for vendor ID
|
|
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
|
|
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
|
|
|
|
# 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_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
|
|
is_active = Column(
|
|
Boolean, default=True
|
|
) # Boolean to indicate if the vendor is active
|
|
is_verified = Column(
|
|
Boolean, default=False
|
|
) # Boolean to indicate if the vendor is verified
|
|
|
|
# ========================================================================
|
|
# Relationships
|
|
# ========================================================================
|
|
owner = relationship(
|
|
"User", back_populates="owned_vendors"
|
|
) # Relationship with User model for the vendor's owner
|
|
vendor_users = relationship(
|
|
"VendorUser", back_populates="vendor"
|
|
) # Relationship with VendorUser model for users in this vendor
|
|
products = relationship(
|
|
"Product", back_populates="vendor"
|
|
) # Relationship with Product model for products of this vendor
|
|
customers = relationship(
|
|
"Customer", back_populates="vendor"
|
|
) # Relationship with Customer model for customers of this vendor
|
|
orders = relationship(
|
|
"Order", back_populates="vendor"
|
|
) # Relationship with Order model for orders placed by this vendor
|
|
marketplace_import_jobs = relationship(
|
|
"MarketplaceImportJob", back_populates="vendor"
|
|
) # Relationship with MarketplaceImportJob model for import jobs related to this vendor
|
|
|
|
domains = relationship(
|
|
"VendorDomain",
|
|
back_populates="vendor",
|
|
cascade="all, delete-orphan",
|
|
order_by="VendorDomain.is_primary.desc()",
|
|
) # Relationship with VendorDomain model for custom domains of the vendor
|
|
|
|
# Single theme relationship (ONE vendor = ONE theme)
|
|
# A vendor has ONE active theme stored in the vendor_themes table.
|
|
# Theme presets available: default, modern, classic, minimal, vibrant
|
|
vendor_theme = relationship(
|
|
"VendorTheme",
|
|
back_populates="vendor",
|
|
uselist=False,
|
|
cascade="all, delete-orphan",
|
|
) # Relationship with VendorTheme model for the active theme of the vendor
|
|
|
|
# Content pages relationship (vendor can override platform default pages)
|
|
content_pages = relationship(
|
|
"ContentPage", back_populates="vendor", cascade="all, delete-orphan"
|
|
) # Relationship with ContentPage model for vendor-specific content pages
|
|
|
|
def __repr__(self):
|
|
"""String representation of the Vendor object."""
|
|
return f"<Vendor(id={self.id}, vendor_code='{self.vendor_code}', name='{self.name}', subdomain='{self.subdomain}')>"
|
|
|
|
# ========================================================================
|
|
# Theme Helper Methods to get active theme and other related information
|
|
# ========================================================================
|
|
|
|
def get_effective_theme(self) -> dict:
|
|
"""
|
|
Get active theme for this vendor.
|
|
|
|
Returns theme from vendor_themes table, or default theme if not set.
|
|
|
|
Returns:
|
|
dict: Theme configuration with colors, fonts, layout, etc.
|
|
"""
|
|
# Check vendor_themes table
|
|
if self.vendor_theme and self.vendor_theme.is_active:
|
|
return self.vendor_theme.to_dict()
|
|
|
|
# Return default theme
|
|
return self._get_default_theme()
|
|
|
|
def _get_default_theme(self) -> dict:
|
|
"""Return the default theme configuration."""
|
|
return {
|
|
"theme_name": "default",
|
|
"colors": {
|
|
"primary": "#6366f1",
|
|
"secondary": "#8b5cf6",
|
|
"accent": "#ec4899",
|
|
"background": "#ffffff",
|
|
"text": "#1f2937",
|
|
"border": "#e5e7eb",
|
|
},
|
|
"fonts": {"heading": "Inter, sans-serif", "body": "Inter, sans-serif"},
|
|
"branding": {
|
|
"logo": None,
|
|
"logo_dark": None,
|
|
"favicon": None,
|
|
"banner": None,
|
|
},
|
|
"layout": {"style": "grid", "header": "fixed", "product_card": "modern"},
|
|
"social_links": {},
|
|
"custom_css": None,
|
|
"css_variables": {
|
|
"--color-primary": "#6366f1",
|
|
"--color-secondary": "#8b5cf6",
|
|
"--color-accent": "#ec4899",
|
|
"--color-background": "#ffffff",
|
|
"--color-text": "#1f2937",
|
|
"--color-border": "#e5e7eb",
|
|
"--font-heading": "Inter, sans-serif",
|
|
"--font-body": "Inter, sans-serif",
|
|
},
|
|
}
|
|
|
|
def get_primary_color(self) -> str:
|
|
"""Get primary color from active theme."""
|
|
theme = self.get_effective_theme()
|
|
return theme.get("colors", {}).get(
|
|
"primary", "#6366f1"
|
|
) # Default to default theme if not found
|
|
|
|
def get_logo_url(self) -> str:
|
|
"""Get logo URL from active theme."""
|
|
theme = self.get_effective_theme()
|
|
return theme.get("branding", {}).get(
|
|
"logo"
|
|
) # Return None or the logo URL if found
|
|
|
|
# ========================================================================
|
|
# Domain Helper Methods
|
|
# ========================================================================
|
|
|
|
@property
|
|
def primary_domain(self):
|
|
"""Get the primary custom domain for this vendor."""
|
|
for domain in self.domains:
|
|
if domain.is_primary and domain.is_active:
|
|
return domain.domain # Return the domain if it's primary and active
|
|
return None
|
|
|
|
@property
|
|
def all_domains(self):
|
|
"""Get all active domains (subdomain + custom domains)."""
|
|
domains = [
|
|
f"{self.subdomain}.{settings.platform_domain}"
|
|
] # Start with the main subdomain
|
|
for domain in self.domains:
|
|
if domain.is_active:
|
|
domains.append(domain.domain) # Add other active custom domains
|
|
return domains
|
|
|
|
|
|
class VendorUserType(str, enum.Enum):
|
|
"""Types of vendor users."""
|
|
|
|
OWNER = "owner" # Vendor owner (full access to vendor area)
|
|
TEAM_MEMBER = "member" # Team member (role-based access to vendor area)
|
|
|
|
|
|
class VendorUser(Base, TimestampMixin):
|
|
"""
|
|
Represents a user's membership in a vendor.
|
|
|
|
- Owner: Created automatically when vendor is created
|
|
- Team Member: Invited by owner via email
|
|
"""
|
|
|
|
__tablename__ = "vendor_users"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
"""Unique identifier for each VendorUser entry."""
|
|
|
|
vendor_id = Column(Integer, ForeignKey("vendors.id"), nullable=False)
|
|
"""Foreign key linking to the associated Vendor."""
|
|
|
|
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
|
|
"""Foreign key linking to the associated User."""
|
|
|
|
# Distinguish between owner and team member
|
|
user_type = Column(String, nullable=False, default=VendorUserType.TEAM_MEMBER.value)
|
|
|
|
# Role for team members (NULL for owners - they have all permissions)
|
|
role_id = Column(Integer, ForeignKey("roles.id"), nullable=True)
|
|
"""Foreign key linking to the associated Role."""
|
|
|
|
invited_by = Column(Integer, ForeignKey("users.id"))
|
|
"""Foreign key linking to the user who invited this VendorUser."""
|
|
invitation_token = Column(String, nullable=True, index=True) # For email activation
|
|
invitation_sent_at = Column(DateTime, nullable=True)
|
|
invitation_accepted_at = Column(DateTime, nullable=True)
|
|
|
|
is_active = Column(
|
|
Boolean, default=False, nullable=False
|
|
) # False until invitation accepted
|
|
"""Indicates whether the VendorUser role is active."""
|
|
|
|
# Relationships
|
|
vendor = relationship("Vendor", back_populates="vendor_users")
|
|
"""Relationship to the Vendor model, representing the associated vendor."""
|
|
|
|
user = relationship(
|
|
"User", foreign_keys=[user_id], back_populates="vendor_memberships"
|
|
)
|
|
"""Relationship to the User model, representing the user who holds this role within the vendor."""
|
|
|
|
inviter = relationship("User", foreign_keys=[invited_by])
|
|
"""Optional relationship to the User model, representing the user who invited this VendorUser."""
|
|
|
|
role = relationship("Role", back_populates="vendor_users")
|
|
"""Relationship to the Role model, representing the role held by the vendor user."""
|
|
|
|
def __repr__(self) -> str:
|
|
"""Return a string representation of the VendorUser instance.
|
|
|
|
Returns:
|
|
str: A string that includes the vendor_id, the user_id and the user_type of the VendorUser instance.
|
|
"""
|
|
return f"<VendorUser(vendor_id={self.vendor_id}, user_id={self.user_id}, type={self.user_type})>"
|
|
|
|
@property
|
|
def is_owner(self) -> bool:
|
|
"""Check if this is an owner membership."""
|
|
return self.user_type == VendorUserType.OWNER.value
|
|
|
|
@property
|
|
def is_team_member(self) -> bool:
|
|
"""Check if this is a team member (not owner)."""
|
|
return self.user_type == VendorUserType.TEAM_MEMBER.value
|
|
|
|
@property
|
|
def is_invitation_pending(self) -> bool:
|
|
"""Check if invitation is still pending."""
|
|
return self.invitation_token is not None and self.invitation_accepted_at is None
|
|
|
|
def has_permission(self, permission: str) -> bool:
|
|
"""
|
|
Check if user has a specific permission.
|
|
|
|
Owners always have all permissions.
|
|
Team members check their role's permissions.
|
|
"""
|
|
# Owners have all permissions
|
|
if self.is_owner:
|
|
return True
|
|
|
|
# Inactive users have no permissions
|
|
if not self.is_active:
|
|
return False
|
|
|
|
# Check role permissions
|
|
if self.role and self.role.permissions:
|
|
return permission in self.role.permissions
|
|
|
|
return False
|
|
|
|
def get_all_permissions(self) -> list:
|
|
"""Get all permissions this user has."""
|
|
if self.is_owner:
|
|
# Return all possible permissions
|
|
from app.core.permissions import VendorPermissions
|
|
|
|
return list(VendorPermissions.__members__.values())
|
|
|
|
if self.role and self.role.permissions:
|
|
return self.role.permissions
|
|
|
|
return []
|
|
|
|
|
|
class Role(Base, TimestampMixin):
|
|
"""Represents a role within a vendor's system."""
|
|
|
|
__tablename__ = "roles" # Name of the table in the database
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
"""Unique identifier for each Role entry."""
|
|
|
|
vendor_id = Column(Integer, ForeignKey("vendors.id"), nullable=False)
|
|
"""Foreign key linking to the associated Vendor."""
|
|
|
|
name = Column(String(100), nullable=False)
|
|
"""Name of the role, with a maximum length of 100 characters."""
|
|
|
|
permissions = Column(JSON, default=list)
|
|
"""Permissions assigned to this role, stored as a JSON array."""
|
|
|
|
vendor = relationship("Vendor")
|
|
"""Relationship to the Vendor model, representing the associated vendor."""
|
|
|
|
vendor_users = relationship("VendorUser", back_populates="role")
|
|
"""Back-relationship to the VendorUser model, representing users with this role."""
|
|
|
|
def __repr__(self) -> str:
|
|
"""Return a string representation of the Role instance.
|
|
|
|
Returns:
|
|
str: A string that includes the id and name of the Role instance.
|
|
"""
|
|
return f"<Role(id={self.id}, name='{self.name}', vendor_id={self.vendor_id})>"
|