cleanup
This commit is contained in:
@@ -27,7 +27,7 @@ from sqlalchemy import (
|
||||
func,
|
||||
)
|
||||
from sqlalchemy import Enum as SQLEnum
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship, validates
|
||||
|
||||
from src.server.database.base import Base, TimestampMixin
|
||||
|
||||
@@ -114,6 +114,58 @@ class AnimeSeries(Base, TimestampMixin):
|
||||
cascade="all, delete-orphan"
|
||||
)
|
||||
|
||||
@validates('key')
|
||||
def validate_key(self, key: str, value: str) -> str:
|
||||
"""Validate key field length and format."""
|
||||
if not value or not value.strip():
|
||||
raise ValueError("Series key cannot be empty")
|
||||
if len(value) > 255:
|
||||
raise ValueError("Series key must be 255 characters or less")
|
||||
return value.strip()
|
||||
|
||||
@validates('name')
|
||||
def validate_name(self, key: str, value: str) -> str:
|
||||
"""Validate name field length."""
|
||||
if not value or not value.strip():
|
||||
raise ValueError("Series name cannot be empty")
|
||||
if len(value) > 500:
|
||||
raise ValueError("Series name must be 500 characters or less")
|
||||
return value.strip()
|
||||
|
||||
@validates('site')
|
||||
def validate_site(self, key: str, value: str) -> str:
|
||||
"""Validate site URL length."""
|
||||
if not value or not value.strip():
|
||||
raise ValueError("Series site URL cannot be empty")
|
||||
if len(value) > 500:
|
||||
raise ValueError("Site URL must be 500 characters or less")
|
||||
return value.strip()
|
||||
|
||||
@validates('folder')
|
||||
def validate_folder(self, key: str, value: str) -> str:
|
||||
"""Validate folder path length."""
|
||||
if not value or not value.strip():
|
||||
raise ValueError("Series folder path cannot be empty")
|
||||
if len(value) > 1000:
|
||||
raise ValueError("Folder path must be 1000 characters or less")
|
||||
return value.strip()
|
||||
|
||||
@validates('cover_url')
|
||||
def validate_cover_url(self, key: str, value: Optional[str]) -> Optional[str]:
|
||||
"""Validate cover URL length."""
|
||||
if value is not None and len(value) > 1000:
|
||||
raise ValueError("Cover URL must be 1000 characters or less")
|
||||
return value
|
||||
|
||||
@validates('total_episodes')
|
||||
def validate_total_episodes(self, key: str, value: Optional[int]) -> Optional[int]:
|
||||
"""Validate total episodes is positive."""
|
||||
if value is not None and value < 0:
|
||||
raise ValueError("Total episodes must be non-negative")
|
||||
if value is not None and value > 10000:
|
||||
raise ValueError("Total episodes must be 10000 or less")
|
||||
return value
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"<AnimeSeries(id={self.id}, key='{self.key}', name='{self.name}')>"
|
||||
|
||||
@@ -190,6 +242,47 @@ class Episode(Base, TimestampMixin):
|
||||
back_populates="episodes"
|
||||
)
|
||||
|
||||
@validates('season')
|
||||
def validate_season(self, key: str, value: int) -> int:
|
||||
"""Validate season number is positive."""
|
||||
if value < 0:
|
||||
raise ValueError("Season number must be non-negative")
|
||||
if value > 1000:
|
||||
raise ValueError("Season number must be 1000 or less")
|
||||
return value
|
||||
|
||||
@validates('episode_number')
|
||||
def validate_episode_number(self, key: str, value: int) -> int:
|
||||
"""Validate episode number is positive."""
|
||||
if value < 0:
|
||||
raise ValueError("Episode number must be non-negative")
|
||||
if value > 10000:
|
||||
raise ValueError("Episode number must be 10000 or less")
|
||||
return value
|
||||
|
||||
@validates('title')
|
||||
def validate_title(self, key: str, value: Optional[str]) -> Optional[str]:
|
||||
"""Validate title length."""
|
||||
if value is not None and len(value) > 500:
|
||||
raise ValueError("Episode title must be 500 characters or less")
|
||||
return value
|
||||
|
||||
@validates('file_path')
|
||||
def validate_file_path(
|
||||
self, key: str, value: Optional[str]
|
||||
) -> Optional[str]:
|
||||
"""Validate file path length."""
|
||||
if value is not None and len(value) > 1000:
|
||||
raise ValueError("File path must be 1000 characters or less")
|
||||
return value
|
||||
|
||||
@validates('file_size')
|
||||
def validate_file_size(self, key: str, value: Optional[int]) -> Optional[int]:
|
||||
"""Validate file size is non-negative."""
|
||||
if value is not None and value < 0:
|
||||
raise ValueError("File size must be non-negative")
|
||||
return value
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return (
|
||||
f"<Episode(id={self.id}, series_id={self.series_id}, "
|
||||
@@ -334,6 +427,87 @@ class DownloadQueueItem(Base, TimestampMixin):
|
||||
back_populates="download_items"
|
||||
)
|
||||
|
||||
@validates('season')
|
||||
def validate_season(self, key: str, value: int) -> int:
|
||||
"""Validate season number is positive."""
|
||||
if value < 0:
|
||||
raise ValueError("Season number must be non-negative")
|
||||
if value > 1000:
|
||||
raise ValueError("Season number must be 1000 or less")
|
||||
return value
|
||||
|
||||
@validates('episode_number')
|
||||
def validate_episode_number(self, key: str, value: int) -> int:
|
||||
"""Validate episode number is positive."""
|
||||
if value < 0:
|
||||
raise ValueError("Episode number must be non-negative")
|
||||
if value > 10000:
|
||||
raise ValueError("Episode number must be 10000 or less")
|
||||
return value
|
||||
|
||||
@validates('progress_percent')
|
||||
def validate_progress_percent(self, key: str, value: float) -> float:
|
||||
"""Validate progress is between 0 and 100."""
|
||||
if value < 0.0:
|
||||
raise ValueError("Progress percent must be non-negative")
|
||||
if value > 100.0:
|
||||
raise ValueError("Progress percent cannot exceed 100")
|
||||
return value
|
||||
|
||||
@validates('downloaded_bytes')
|
||||
def validate_downloaded_bytes(self, key: str, value: int) -> int:
|
||||
"""Validate downloaded bytes is non-negative."""
|
||||
if value < 0:
|
||||
raise ValueError("Downloaded bytes must be non-negative")
|
||||
return value
|
||||
|
||||
@validates('total_bytes')
|
||||
def validate_total_bytes(
|
||||
self, key: str, value: Optional[int]
|
||||
) -> Optional[int]:
|
||||
"""Validate total bytes is non-negative."""
|
||||
if value is not None and value < 0:
|
||||
raise ValueError("Total bytes must be non-negative")
|
||||
return value
|
||||
|
||||
@validates('download_speed')
|
||||
def validate_download_speed(
|
||||
self, key: str, value: Optional[float]
|
||||
) -> Optional[float]:
|
||||
"""Validate download speed is non-negative."""
|
||||
if value is not None and value < 0.0:
|
||||
raise ValueError("Download speed must be non-negative")
|
||||
return value
|
||||
|
||||
@validates('retry_count')
|
||||
def validate_retry_count(self, key: str, value: int) -> int:
|
||||
"""Validate retry count is non-negative."""
|
||||
if value < 0:
|
||||
raise ValueError("Retry count must be non-negative")
|
||||
if value > 100:
|
||||
raise ValueError("Retry count cannot exceed 100")
|
||||
return value
|
||||
|
||||
@validates('download_url')
|
||||
def validate_download_url(
|
||||
self, key: str, value: Optional[str]
|
||||
) -> Optional[str]:
|
||||
"""Validate download URL length."""
|
||||
if value is not None and len(value) > 1000:
|
||||
raise ValueError("Download URL must be 1000 characters or less")
|
||||
return value
|
||||
|
||||
@validates('file_destination')
|
||||
def validate_file_destination(
|
||||
self, key: str, value: Optional[str]
|
||||
) -> Optional[str]:
|
||||
"""Validate file destination path length."""
|
||||
if value is not None and len(value) > 1000:
|
||||
raise ValueError(
|
||||
"File destination path must be 1000 characters or less"
|
||||
)
|
||||
return value
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return (
|
||||
f"<DownloadQueueItem(id={self.id}, "
|
||||
@@ -412,6 +586,51 @@ class UserSession(Base, TimestampMixin):
|
||||
doc="Last activity timestamp"
|
||||
)
|
||||
|
||||
@validates('session_id')
|
||||
def validate_session_id(self, key: str, value: str) -> str:
|
||||
"""Validate session ID length and format."""
|
||||
if not value or not value.strip():
|
||||
raise ValueError("Session ID cannot be empty")
|
||||
if len(value) > 255:
|
||||
raise ValueError("Session ID must be 255 characters or less")
|
||||
return value.strip()
|
||||
|
||||
@validates('token_hash')
|
||||
def validate_token_hash(self, key: str, value: str) -> str:
|
||||
"""Validate token hash length."""
|
||||
if not value or not value.strip():
|
||||
raise ValueError("Token hash cannot be empty")
|
||||
if len(value) > 255:
|
||||
raise ValueError("Token hash must be 255 characters or less")
|
||||
return value.strip()
|
||||
|
||||
@validates('user_id')
|
||||
def validate_user_id(
|
||||
self, key: str, value: Optional[str]
|
||||
) -> Optional[str]:
|
||||
"""Validate user ID length."""
|
||||
if value is not None and len(value) > 255:
|
||||
raise ValueError("User ID must be 255 characters or less")
|
||||
return value
|
||||
|
||||
@validates('ip_address')
|
||||
def validate_ip_address(
|
||||
self, key: str, value: Optional[str]
|
||||
) -> Optional[str]:
|
||||
"""Validate IP address length (IPv4 or IPv6)."""
|
||||
if value is not None and len(value) > 45:
|
||||
raise ValueError("IP address must be 45 characters or less")
|
||||
return value
|
||||
|
||||
@validates('user_agent')
|
||||
def validate_user_agent(
|
||||
self, key: str, value: Optional[str]
|
||||
) -> Optional[str]:
|
||||
"""Validate user agent length."""
|
||||
if value is not None and len(value) > 500:
|
||||
raise ValueError("User agent must be 500 characters or less")
|
||||
return value
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return (
|
||||
f"<UserSession(id={self.id}, "
|
||||
|
||||
Reference in New Issue
Block a user