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

Vendor team member management features

Browse Source
This commit is contained in:
Samir Boulahtit
2025-11-14 21:08:57 +01:00
parent af23f5b88f
commit 41439eed09
10 changed files with 1786 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

115
models/database/user.py
View File

@@ -1,39 +1,56 @@
# models/database/user.py
# models/database/user.py - IMPROVED VERSION
"""
User model with authentication support.
This module defines the User model which includes fields for user details,
authentication information, and relationships to other models such as Vendor and Customer.
"""
from sqlalchemy import Boolean, Column, DateTime, Integer, String
from sqlalchemy.orm import relationship
ROLE CLARIFICATION:
- User.role should ONLY contain platform-level roles:
* "admin" - Platform administrator (full system access)
* "vendor" - Any user who owns or is part of a vendor team
- Vendor-specific roles (manager, staff, etc.) are stored in VendorUser.role
- Customers are NOT in the User table - they use the Customer model
"""
from sqlalchemy import Boolean, Column, DateTime, Integer, String, Enum
from sqlalchemy.orm import relationship
import enum
# 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 UserRole(str, enum.Enum):
"""Platform-level user roles."""
ADMIN = "admin" # Platform administrator
VENDOR = "vendor" # Vendor owner or team member
class User(Base, TimestampMixin):
"""Represents a user in the system."""
"""Represents a platform user (admins and vendors only)."""
__tablename__ = "users" # Name of the table in the database
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True) # Primary key and indexed column for user ID
email = Column(String, unique=True, index=True, nullable=False) # Unique, indexed, non-nullable email column
username = Column(String, unique=True, index=True, nullable=False) # Unique, indexed, non-nullable username column
first_name = Column(String) # Optional first name column
last_name = Column(String) # Optional last name column
hashed_password = Column(String, nullable=False) # Non-nullable hashed password column
role = Column(String, nullable=False, default="user") # Role of the user (default is 'user') //TODO: Change to customer, vendor, admin
is_active = Column(Boolean, default=True, nullable=False) # Active status of the user (default is True)
last_login = Column(DateTime, nullable=True) # Optional last login timestamp column
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
username = Column(String, unique=True, index=True, nullable=False)
first_name = Column(String)
last_name = Column(String)
hashed_password = Column(String, nullable=False)
# Platform-level role only (admin or vendor)
role = Column(String, nullable=False, default=UserRole.VENDOR.value)
is_active = Column(Boolean, default=True, nullable=False)
is_email_verified = Column(Boolean, default=False, nullable=False)
last_login = Column(DateTime, nullable=True)
# Relationships
marketplace_import_jobs = relationship("MarketplaceImportJob",
back_populates="user") # Relationship with import jobs
owned_vendors = relationship("Vendor", back_populates="owner") # Relationship with vendors owned by this user
vendor_memberships = relationship("VendorUser", foreign_keys="[VendorUser.user_id]",
back_populates="user") # Relationship with vendor memberships
marketplace_import_jobs = relationship("MarketplaceImportJob", back_populates="user")
owned_vendors = relationship("Vendor", back_populates="owner")
vendor_memberships = relationship(
"VendorUser",
foreign_keys="[VendorUser.user_id]",
back_populates="user"
)
def __repr__(self):
"""String representation of the User object."""
@@ -45,3 +62,55 @@ class User(Base, TimestampMixin):
if self.first_name and self.last_name:
return f"{self.first_name} {self.last_name}"
return self.username
@property
def is_admin(self) -> bool:
"""Check if user is a platform admin."""
return self.role == UserRole.ADMIN.value
@property
def is_vendor(self) -> bool:
"""Check if user is a vendor (owner or team member)."""
return self.role == UserRole.VENDOR.value
def is_owner_of(self, vendor_id: int) -> bool:
"""Check if user is the owner of a specific vendor."""
return any(v.id == vendor_id for v in self.owned_vendors)
def is_member_of(self, vendor_id: int) -> bool:
"""Check if user is a member of a specific vendor (owner or team)."""
# Check if owner
if self.is_owner_of(vendor_id):
return True
# Check if team member
return any(
vm.vendor_id == vendor_id and vm.is_active
for vm in self.vendor_memberships
)
def get_vendor_role(self, vendor_id: int) -> str:
"""Get user's role within a specific vendor."""
# Check if owner
if self.is_owner_of(vendor_id):
return "owner"
# Check team membership
for vm in self.vendor_memberships:
if vm.vendor_id == vendor_id and vm.is_active:
return vm.role.name if vm.role else "member"
return None
def has_vendor_permission(self, vendor_id: int, permission: str) -> bool:
"""Check if user has a specific permission in a vendor."""
# Owners have all permissions
if self.is_owner_of(vendor_id):
return True
# Check team member permissions
for vm in self.vendor_memberships:
if vm.vendor_id == vendor_id and vm.is_active:
if vm.role and permission in vm.role.permissions:
return True
return False

84
models/database/vendor.py
View File

@@ -5,14 +5,14 @@ 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.
"""
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, Text, JSON
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, Text, JSON, DateTime
from sqlalchemy.orm import relationship
# 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
from app.core.config import settings
import enum
class Vendor(Base, TimestampMixin):
"""Represents a vendor in the system."""
@@ -177,10 +177,21 @@ class Vendor(Base, TimestampMixin):
return domains
class VendorUser(Base, TimestampMixin):
"""Represents a user's role within a specific vendor."""
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)
__tablename__ = "vendor_users" # Name of the table in the database
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."""
@@ -191,15 +202,23 @@ class VendorUser(Base, TimestampMixin):
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
"""Foreign key linking to the associated User."""
role_id = Column(Integer, ForeignKey("roles.id"), nullable=False)
# 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=True, nullable=False)
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."""
@@ -216,10 +235,57 @@ class VendorUser(Base, TimestampMixin):
"""Return a string representation of the VendorUser instance.
Returns:
str: A string that includes the vendor_id and user_id of the VendorUser instance.
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})>"
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."""

261
models/schema/team.py
View File

@@ -1 +1,260 @@
# Team management models
# models/schema/team.py
"""
Pydantic schemas for vendor team management.
This module defines request/response schemas for:
- Team member listing
- Team member invitation
- Team member updates
- Role management
"""
from datetime import datetime
from typing import Optional, List
from pydantic import BaseModel, EmailStr, Field, validator
# ============================================================================
# Role Schemas
# ============================================================================
class RoleBase(BaseModel):
"""Base role schema."""
name: str = Field(..., min_length=1, max_length=100, description="Role name")
permissions: List[str] = Field(default_factory=list, description="List of permission strings")
class RoleCreate(RoleBase):
"""Schema for creating a role."""
pass
class RoleUpdate(BaseModel):
"""Schema for updating a role."""
name: Optional[str] = Field(None, min_length=1, max_length=100)
permissions: Optional[List[str]] = None
class RoleResponse(RoleBase):
"""Schema for role response."""
id: int
vendor_id: int
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True # Pydantic v2 (use orm_mode = True for v1)
class RoleListResponse(BaseModel):
"""Schema for role list response."""
roles: List[RoleResponse]
total: int
# ============================================================================
# Team Member Schemas
# ============================================================================
class TeamMemberBase(BaseModel):
"""Base team member schema."""
email: EmailStr = Field(..., description="Team member email address")
first_name: Optional[str] = Field(None, max_length=100)
last_name: Optional[str] = Field(None, max_length=100)
class TeamMemberInvite(TeamMemberBase):
"""Schema for inviting a team member."""
role_id: Optional[int] = Field(None, description="Role ID to assign (for preset roles)")
role_name: Optional[str] = Field(None, description="Role name (manager, staff, support, etc.)")
custom_permissions: Optional[List[str]] = Field(
None,
description="Custom permissions (overrides role preset)"
)
@validator('role_name')
def validate_role_name(cls, v):
"""Validate role name is in allowed presets."""
if v is not None:
allowed_roles = ['manager', 'staff', 'support', 'viewer', 'marketing']
if v.lower() not in allowed_roles:
raise ValueError(
f"Role name must be one of: {', '.join(allowed_roles)}"
)
return v.lower() if v else v
@validator('custom_permissions')
def validate_custom_permissions(cls, v, values):
"""Ensure either role_id/role_name OR custom_permissions is provided."""
if v is not None and len(v) > 0:
# If custom permissions provided, role_name should be provided too
if 'role_name' not in values or not values['role_name']:
raise ValueError(
"role_name is required when providing custom_permissions"
)
return v
class TeamMemberUpdate(BaseModel):
"""Schema for updating a team member."""
role_id: Optional[int] = Field(None, description="New role ID")
is_active: Optional[bool] = Field(None, description="Active status")
class TeamMemberResponse(BaseModel):
"""Schema for team member response."""
id: int = Field(..., description="User ID")
email: EmailStr
username: str
first_name: Optional[str]
last_name: Optional[str]
full_name: str
user_type: str = Field(..., description="'owner' or 'member'")
role_name: str = Field(..., description="Role name")
role_id: Optional[int]
permissions: List[str] = Field(default_factory=list, description="User's permissions")
is_active: bool
is_owner: bool
invitation_pending: bool = Field(
default=False,
description="True if invitation not yet accepted"
)
invited_at: Optional[datetime] = Field(None, description="When invitation was sent")
accepted_at: Optional[datetime] = Field(None, description="When invitation was accepted")
joined_at: datetime = Field(..., description="When user joined vendor")
class Config:
from_attributes = True
class TeamMemberListResponse(BaseModel):
"""Schema for team member list response."""
members: List[TeamMemberResponse]
total: int
active_count: int
pending_invitations: int
# ============================================================================
# Invitation Schemas
# ============================================================================
class InvitationAccept(BaseModel):
"""Schema for accepting a team invitation."""
invitation_token: str = Field(..., min_length=32, description="Invitation token from email")
password: str = Field(
...,
min_length=8,
max_length=128,
description="Password for new account"
)
first_name: str = Field(..., min_length=1, max_length=100)
last_name: str = Field(..., min_length=1, max_length=100)
@validator('password')
def validate_password_strength(cls, v):
"""Validate password meets minimum requirements."""
if len(v) < 8:
raise ValueError("Password must be at least 8 characters long")
has_upper = any(c.isupper() for c in v)
has_lower = any(c.islower() for c in v)
has_digit = any(c.isdigit() for c in v)
if not (has_upper and has_lower and has_digit):
raise ValueError(
"Password must contain at least one uppercase letter, "
"one lowercase letter, and one digit"
)
return v
class InvitationResponse(BaseModel):
"""Schema for invitation response."""
message: str
email: EmailStr
role: str
invitation_token: Optional[str] = Field(
None,
description="Token (only returned in dev/test environments)"
)
invitation_sent: bool = Field(default=True)
class InvitationAcceptResponse(BaseModel):
"""Schema for invitation acceptance response."""
message: str
vendor: dict = Field(..., description="Vendor information")
user: dict = Field(..., description="User information")
role: str
# ============================================================================
# Team Statistics Schema
# ============================================================================
class TeamStatistics(BaseModel):
"""Schema for team statistics."""
total_members: int
active_members: int
inactive_members: int
pending_invitations: int
owners: int
team_members: int
roles_breakdown: dict = Field(
default_factory=dict,
description="Count of members per role"
)
# ============================================================================
# Bulk Operations Schemas
# ============================================================================
class BulkRemoveRequest(BaseModel):
"""Schema for bulk removing team members."""
user_ids: List[int] = Field(..., min_items=1, description="List of user IDs to remove")
class BulkRemoveResponse(BaseModel):
"""Schema for bulk remove response."""
success_count: int
failed_count: int
errors: List[dict] = Field(default_factory=list)
# ============================================================================
# Permission Check Schemas
# ============================================================================
class PermissionCheckRequest(BaseModel):
"""Schema for checking permissions."""
permissions: List[str] = Field(..., min_items=1, description="Permissions to check")
class PermissionCheckResponse(BaseModel):
"""Schema for permission check response."""
has_all: bool = Field(..., description="True if user has all permissions")
has_any: bool = Field(..., description="True if user has any permission")
granted: List[str] = Field(default_factory=list, description="Permissions user has")
denied: List[str] = Field(default_factory=list, description="Permissions user lacks")
class UserPermissionsResponse(BaseModel):
"""Schema for user's permissions response."""
permissions: List[str] = Field(default_factory=list)
permission_count: int
is_owner: bool
role_name: Optional[str] = None
# ============================================================================
# Error Response Schema
# ============================================================================
class TeamErrorResponse(BaseModel):
"""Schema for team operation errors."""
error_code: str
message: str
details: Optional[dict] = None