🔹 Простой Transaction Script
# user_script.py
def create_user(name: str, email: str, password: str):
# Валидация
if not name:
raise ValueError("Name is required")
if "@" not in email:
raise ValueError("Invalid email")
if len(password) < 8:
raise ValueError("Password must be at least 8 characters")
# Проверка существования
existing = db.fetch_one("SELECT id FROM users WHERE email = ?", (email,))
if existing:
raise ValueError("User already exists")
# Создание
user_id = db.execute(
"INSERT INTO users (name, email, password_hash) VALUES (?, ?, ?) RETURNING id",
(name, email, hash_password(password))
)
return {"user_id": user_id, "status": "created"}
def get_user(user_id: int):
row = db.fetch_one("SELECT * FROM users WHERE id = ?", (user_id,))
if not row:
raise ValueError("User not found")
return {
"id": row['id'],
"name": row['name'],
"email": row['email']
}
🔹 Transaction Script с классами
class UserService:
def create_user(self, name: str, email: str, password: str):
# Валидация
self._validate_user(name, email, password)
# Проверка существования
if self._user_exists(email):
raise ValueError("User already exists")
# Создание
user_id = self._save_user(name, email, password)
# Побочные эффекты
self._send_welcome_email(email)
return {"user_id": user_id, "status": "created"}
def _validate_user(self, name: str, email: str, password: str):
if not name:
raise ValueError("Name is required")
if "@" not in email:
raise ValueError("Invalid email")
if len(password) < 8:
raise ValueError("Password must be at least 8 characters")
def _user_exists(self, email: str) -> bool:
row = db.fetch_one("SELECT id FROM users WHERE email = ?", (email,))
return row is not None
def _save_user(self, name: str, email: str, password: str) -> int:
return db.execute(
"INSERT INTO users (name, email, password_hash) VALUES (?, ?, ?) RETURNING id",
(name, email, hash_password(password))
)
def _send_welcome_email(self, email: str):
email_service.send(email, "Welcome!", "Welcome to our service!")