Coverage for backend / app / core / models.py: 96%
114 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-17 21:34 +0000
1"""Database models for application settings and user management."""
3import datetime as dt
4import math
5from typing import Any
7from sqlalchemy import Column, Integer, String, Boolean, TIMESTAMP, CheckConstraint, event
8from sqlalchemy.ext.hybrid import hybrid_property
9from sqlalchemy.orm import relationship
10from sqlalchemy.sql import expression
12from app.base_models import CommonBase, Owned
13from app.config import settings
14from app.database import Base
17class Setting(CommonBase, Base):
18 """Represents the application settings
20 Attributes:
21 -----------
22 - `name` (str, unique): The name of the setting.
23 - `value` (float): The value of the setting.
24 - `description` (str): A description of the setting.
25 - `is_active` (bool): Indicates whether the setting is active."""
27 name = Column(String, nullable=False, unique=True)
28 value = Column(String, nullable=False)
29 description = Column(String, nullable=True)
30 is_active = Column(Boolean, nullable=False, server_default=expression.true())
33def get_setting_value(db, name: str, default: Any):
34 """Retrieve a setting value from the database by its name.
35 :param db: Database session.
36 :param name: The name of the setting to retrieve.
37 :param default: The default value to return if the setting is not found."""
39 entry = db.query(Setting).filter(Setting.name == name).filter(Setting.is_active).first()
40 if entry:
41 return entry.value
42 else:
43 return default
46class User(CommonBase, Base):
47 """Represents core user identification and authentication.
49 Attributes:
50 -----------
51 - `password` (str): Encrypted password for authentication.
52 - `email` (str, unique): User's email address.
53 - `is_active` (bool): Indicates whether the user account is active.
54 - `is_admin` (bool): Indicates whether the user is an administrator.
55 - `is_demo` (bool): Indicates whether the user is a demo account.
56 - `is_verified` (bool): Indicates whether the user's email is verified.
57 - `last_login` (datetime, optional): The timestamp of the last login.
58 - `previous_login` (datetime, optional): The timestamp of the previous login.
59 - `app_version` (str, optional): Version of the application used for the last login.
60 - `first_name` (str, optional): User's first name.
61 - `last_name` (str, optional): User's last name.
62 - `token_version` (int): Version of the token for invalidation purposes.
63 - `name` (str, optional): Computed property that combines first and last name.
65 Relationships:
66 --------------
67 - `preferences` (UserPreferences): One-to-one relationship to user preferences.
68 - `stripe_details` (StripeDetails): One-to-one relationship to Stripe payment details.
69 - `premium` (PremiumSettings): One-to-one relationship to premium subscription settings.
70 - `tokens` (list of UserToken): One-to-many relationship to user tokens."""
72 password = Column(String, nullable=False)
73 email = Column(String, nullable=False, unique=True)
74 is_active = Column(Boolean, nullable=False, server_default=expression.true())
75 is_admin = Column(Boolean, nullable=False, server_default=expression.false())
76 is_demo = Column(Boolean, nullable=False, server_default=expression.false())
77 is_verified = Column(Boolean, nullable=False, server_default=expression.false())
78 last_login = Column(TIMESTAMP(timezone=True), nullable=True)
79 previous_login = Column(TIMESTAMP(timezone=True), nullable=True)
80 app_version = Column(String, nullable=True)
81 first_name = Column(String, nullable=True)
82 last_name = Column(String, nullable=True)
83 token_version = Column(Integer, default=0, nullable=False)
85 # Relationships
86 preferences = relationship("UserPreferences", uselist=False, cascade="all, delete-orphan", lazy="joined")
87 stripe_details = relationship("StripeDetails", uselist=False, cascade="all, delete-orphan", lazy="joined")
88 premium = relationship("PremiumSettings", uselist=False, cascade="all, delete-orphan", lazy="joined")
89 tokens = relationship("UserToken", cascade="all, delete-orphan", lazy="dynamic")
91 def __init__(self, **kwargs) -> None:
92 """Initialise User with automatic creation of related records."""
94 # Extract relationship data before calling super().__init__
95 preferences_data = kwargs.pop("preferences", None)
96 stripe_details_data = kwargs.pop("stripe_details", None)
97 premium_data = kwargs.pop("premium", None)
99 # Call parent constructor with remaining kwargs
100 super().__init__(**kwargs)
102 # Handle preferences - create an instance if dict provided or if not already set
103 if preferences_data:
104 if isinstance(preferences_data, dict):
105 # noinspection PyArgumentList
106 self.preferences = UserPreferences(**preferences_data)
107 else:
108 self.preferences = preferences_data
109 elif not self.preferences:
110 self.preferences = UserPreferences()
112 # Handle stripe_details - create an instance if dict provided or if not already set
113 if stripe_details_data:
114 if isinstance(stripe_details_data, dict):
115 # noinspection PyArgumentList
116 self.stripe_details = StripeDetails(**stripe_details_data)
117 else:
118 self.stripe_details = stripe_details_data
119 elif not self.stripe_details:
120 self.stripe_details = StripeDetails()
122 # Handle premium - create instance if dict provided or if not already set
123 if premium_data:
124 if isinstance(premium_data, dict):
125 # noinspection PyArgumentList
126 self.premium = PremiumSettings(**premium_data)
127 else:
128 self.premium = premium_data
129 elif not self.premium:
130 self.premium = PremiumSettings()
132 @hybrid_property
133 def name(self) -> str | None:
134 """Computed property that combines the first and last name"""
136 if self.first_name and self.last_name:
137 return f"{self.first_name} {self.last_name}"
138 return None
140 @hybrid_property
141 def pending_email_change(self) -> str | None:
142 """Check if there is a pending email change token"""
144 for token in self.tokens:
145 if token.token_type == "email_change":
146 return token.pending_email
147 return None
150class UserPreferences(Owned, Base):
151 """User-specific preferences and settings.
153 Attributes:
154 -----------
155 - `theme` (str): The theme of the application.
156 - `dark_mode` (bool): Indicates whether dark mode is enabled.
157 - `chase_threshold` (int): The threshold for chasing jobs in the dashboard.
158 - `deadline_threshold` (int): The threshold for deadlines in the dashboard.
159 - `update_limit` (int): Max number updates displayed in the dashboard.
160 - `default_currency` (str): The default currency for salary fields."""
162 theme = Column(String, nullable=False, server_default="mixed-berry")
163 dark_mode = Column(String, nullable=False, server_default="system")
164 chase_threshold = Column(Integer, nullable=False, server_default="14")
165 deadline_threshold = Column(Integer, nullable=False, server_default="7")
166 update_limit = Column(Integer, nullable=False, server_default="10")
167 default_currency = Column(String, nullable=False, server_default="GBP")
168 extension_banner_dismissed = Column(Boolean, nullable=False, server_default="false")
171class StripeDetails(Owned, Base):
172 """Stripe payment and subscription information.
174 Attributes:
175 -----------
176 - `customer_id` (str, optional): Stripe customer identifier.
177 - `subscription_id` (str, optional): Stripe subscription identifier.
178 - `subscription_status` (str, optional): Current subscription status.
179 - `trial_end_date` (int, optional): Timestamp of trial end date in seconds since epoch."""
181 customer_id = Column(String, nullable=True)
182 subscription_id = Column(String, nullable=True)
183 subscription_status = Column(String, nullable=True)
184 trial_end_date = Column(Integer, nullable=True)
187class PremiumSettings(Owned, Base):
188 """Premium subscription settings and feature flags.
190 Attributes:
191 -----------
192 - `is_active` (bool): Indicates whether the user has an active premium subscription.
193 - `job_scraping_active` (bool): Indicates whether job scraping is enabled.
194 - `job_rating_active` (bool): Indicates whether job rating is enabled."""
196 is_active = Column(Boolean, nullable=False, server_default=expression.false())
197 job_scraping_active = Column(Boolean, nullable=False, server_default=expression.true())
198 job_rating_active = Column(Boolean, nullable=False, server_default=expression.true())
201class UserToken(Owned, Base):
202 """Authentication and verification tokens.
204 Attributes:
205 -----------
206 - `token` (str, unique): The actual token string.
207 - `token_type` (str): Type of token (verification, password_reset, email_change).
208 - `pending_email` (str, optional): For email_change tokens, the new email address.
209 - `is_valid` (bool): Computed property to check if the token is valid."""
211 token = Column(String, nullable=False, unique=True, index=True)
212 token_type = Column(String, nullable=False)
213 pending_email = Column(String, nullable=True)
215 @hybrid_property
216 def is_valid(self) -> bool:
217 """Check if the token is valid"""
219 # Define expiration times based on the token type
220 expiration_minutes = {
221 "verification": settings.verification_token_expiration_minutes,
222 "password_reset": settings.password_reset_token_expiration_minutes,
223 "email_change": settings.email_change_token_expiration_minutes,
224 }
226 # noinspection PyTypeChecker
227 minutes = expiration_minutes.get(self.token_type, settings.verification_token_expiration_minutes)
228 expiration_time = self.created_at + dt.timedelta(minutes=minutes)
229 return dt.datetime.now(dt.timezone.utc) < expiration_time
231 @hybrid_property
232 def remaining_seconds(self) -> int:
233 """Calculate how many seconds remain until the next email can be sent.
234 :return: seconds remaining until next email can be sent"""
236 time_since_last_email = int((dt.datetime.now(dt.timezone.utc) - self.created_at).total_seconds())
237 return math.ceil(settings.verification_email_min_interval_seconds - time_since_last_email)
240@event.listens_for(UserToken, "before_insert")
241def delete_existing_tokens_of_same_type(mapper, connection, target):
242 """Delete existing tokens of the same type for the same user before inserting a new one."""
244 _ = mapper
245 connection.execute(
246 UserToken.__table__.delete().where(
247 (UserToken.owner_id == target.owner_id) & (UserToken.token_type == target.token_type)
248 )
249 )
252class UserQualification(Owned, Base):
253 """User qualifications for job matching
255 Attributes:
256 -----------
257 - `experience` (str): User's experience details.
258 - `skills` (str): User's skills details.
259 - `qualities` (str): User's personal qualities.
260 - `education` (str): User's education details.
261 - `interests` (str): User's job interests.
263 Relationships:
264 --------------
265 - `job_ratings` (list of JobRating): List of job ratings associated with the user qualification.
267 Constraints
268 ------------
269 - At least one of experience, skills, qualities, education, or interests must be provided"""
271 experience = Column(String, nullable=True)
272 skills = Column(String, nullable=True)
273 qualities = Column(String, nullable=True)
274 education = Column(String, nullable=True)
275 interests = Column(String, nullable=True)
277 job_ratings = relationship("JobRating", back_populates="user_qualification")
279 __table_args__ = (
280 CheckConstraint(
281 "experience IS NOT NULL OR skills IS NOT NULL OR qualities IS NOT NULL OR education IS NOT NULL OR interests IS NOT NULL",
282 name="user_qualification_data_required",
283 ),
284 )