-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathmodels_email.py
More file actions
58 lines (48 loc) · 2.18 KB
/
models_email.py
File metadata and controls
58 lines (48 loc) · 2.18 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
"""
Email Database Models - Separate database for email data isolation
"""
import os
import sys
from datetime import datetime
from sqlmodel import Field, SQLModel, create_engine, Session, MetaData
import uuid
# Create separate metadata for email tables to avoid importing all main DB tables
email_metadata = MetaData()
class UserEmail(SQLModel, table=True, metadata=email_metadata):
"""User email associations stored in separate database for security isolation"""
__tablename__ = "user_email"
id: str = Field(default_factory=lambda: uuid.uuid4().hex, primary_key=True)
user_id: str = Field(index=True) # Links to User.display_name in main db (no FK for isolation)
email_encrypted: str = Field(index=True) # AES encrypted email address
email_verified: bool = Field(default=False)
verification_token: str | None = Field(default=None, index=True)
added_at: datetime = Field(default_factory=datetime.utcnow)
verified_at: datetime | None = Field(default=None)
def __repr__(self):
return f"<UserEmail(user_id={self.user_id}, verified={self.email_verified})>"
# Database path configuration for email database
def get_email_database_path():
"""
Email database path selection with test environment safety
"""
# Safety: Detect test environment
if ('pytest' in sys.modules or
'PYTEST_CURRENT_TEST' in os.environ or
any('pytest' in arg for arg in sys.argv)):
return ':memory:'
# Explicit configuration
if 'EMAIL_DATABASE_PATH' in os.environ and os.environ['EMAIL_DATABASE_PATH'].strip():
return os.environ['EMAIL_DATABASE_PATH']
# Default: separate email database
return 'email.db'
# Get email database path and create engine
EMAIL_DATABASE_PATH = get_email_database_path()
print(f"Email database path: {EMAIL_DATABASE_PATH}")
email_engine = create_engine(
f"sqlite:///{EMAIL_DATABASE_PATH}" if EMAIL_DATABASE_PATH != ':memory:' else "sqlite:///:memory:",
echo=False,
connect_args={"check_same_thread": False},
pool_pre_ping=True if EMAIL_DATABASE_PATH != ':memory:' else False
)
# Create only email tables in the email database
UserEmail.__table__.create(email_engine, checkfirst=True)