fix: address database model quality issues

- Fix mutable default arguments with lambda
- Add indexes on foreign keys
- Replace deprecated datetime.utcnow
- Add __repr__ methods for debugging
- Remove redundant primary key indexes
This commit is contained in:
2026-05-11 20:46:03 +02:00
parent 13488f7b21
commit caf5bcb172
+85 -35
View File
@@ -1,82 +1,132 @@
from sqlalchemy import Column, Integer, String, DateTime, Boolean, Text, ForeignKey, JSON from sqlalchemy import Column, Integer, String, DateTime, Boolean, Text, ForeignKey, JSON
from sqlalchemy.orm import relationship from sqlalchemy.orm import relationship
from datetime import datetime from datetime import datetime, timezone
from .database import Base from .database import Base
class Source(Base): class Source(Base):
__tablename__ = "sources" __tablename__ = "sources"
id = Column(Integer, primary_key=True, index=True) id = Column(Integer, primary_key=True)
name = Column(String, nullable=False) name = Column(String, nullable=False)
type = Column(String, nullable=False) # local, ssh, database type = Column(String, nullable=False) # local, ssh, database
config = Column(JSON, default=dict) config = Column(JSON, default=lambda: {})
created_at = Column(DateTime, default=datetime.utcnow) created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) updated_at = Column(
DateTime,
default=lambda: datetime.now(timezone.utc),
onupdate=lambda: datetime.now(timezone.utc),
)
jobs = relationship("Job", back_populates="source", cascade="all, delete-orphan") jobs = relationship("Job", back_populates="source", cascade="all, delete-orphan")
def __repr__(self):
return f"<Source(id={self.id}, name='{self.name}', type='{self.type}')>"
class Job(Base): class Job(Base):
__tablename__ = "jobs" __tablename__ = "jobs"
id = Column(Integer, primary_key=True, index=True) id = Column(Integer, primary_key=True)
name = Column(String, nullable=False) name = Column(String, nullable=False)
source_id = Column(Integer, ForeignKey("sources.id"), nullable=False) source_id = Column(Integer, ForeignKey("sources.id"), nullable=False, index=True)
strategy = Column(String, nullable=False, default="full") # full, incremental strategy = Column(String, nullable=False, default="full") # full, incremental
destination_path = Column(String, nullable=False) destination_path = Column(String, nullable=False)
exclude_patterns = Column(JSON, default=list) exclude_patterns = Column(JSON, default=lambda: [])
enabled = Column(Boolean, default=True) enabled = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow) created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) updated_at = Column(
DateTime,
default=lambda: datetime.now(timezone.utc),
onupdate=lambda: datetime.now(timezone.utc),
)
source = relationship("Source", back_populates="jobs") source = relationship("Source", back_populates="jobs")
schedule = relationship("Schedule", back_populates="job", uselist=False, cascade="all, delete-orphan") schedule = relationship(
executions = relationship("JobExecution", back_populates="job", cascade="all, delete-orphan") "Schedule", back_populates="job", uselist=False, cascade="all, delete-orphan"
)
executions = relationship(
"JobExecution", back_populates="job", cascade="all, delete-orphan"
)
def __repr__(self):
return f"<Job(id={self.id}, name='{self.name}', source_id={self.source_id})>"
class Schedule(Base): class Schedule(Base):
__tablename__ = "schedules" __tablename__ = "schedules"
id = Column(Integer, primary_key=True, index=True) id = Column(Integer, primary_key=True)
job_id = Column(Integer, ForeignKey("jobs.id"), unique=True, nullable=False) job_id = Column(
Integer, ForeignKey("jobs.id"), unique=True, nullable=False, index=True
)
cron_expression = Column(String, nullable=False) cron_expression = Column(String, nullable=False)
enabled = Column(Boolean, default=True) enabled = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow) created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
job = relationship("Job", back_populates="schedule") job = relationship("Job", back_populates="schedule")
def __repr__(self):
return f"<Schedule(id={self.id}, job_id={self.job_id}, cron='{self.cron_expression}')>"
class JobExecution(Base): class JobExecution(Base):
__tablename__ = "job_executions" __tablename__ = "job_executions"
id = Column(Integer, primary_key=True, index=True) id = Column(Integer, primary_key=True)
job_id = Column(Integer, ForeignKey("jobs.id"), nullable=False) job_id = Column(Integer, ForeignKey("jobs.id"), nullable=False, index=True)
status = Column(String, nullable=False, default="pending") # pending, running, success, failed, cancelled status = Column(
String, nullable=False, default="pending"
) # pending, running, success, failed, cancelled
started_at = Column(DateTime, nullable=True) started_at = Column(DateTime, nullable=True)
completed_at = Column(DateTime, nullable=True) completed_at = Column(DateTime, nullable=True)
bytes_processed = Column(Integer, default=0) bytes_processed = Column(Integer, default=0)
bytes_backed_up = Column(Integer, default=0) bytes_backed_up = Column(Integer, default=0)
error_message = Column(Text, nullable=True) error_message = Column(Text, nullable=True)
triggered_by = Column(String, nullable=False) # manual, schedule triggered_by = Column(String, nullable=False) # manual, schedule
job = relationship("Job", back_populates="executions") job = relationship("Job", back_populates="executions")
backups = relationship("Backup", back_populates="execution", cascade="all, delete-orphan") backups = relationship(
"Backup", back_populates="execution", cascade="all, delete-orphan"
)
def __repr__(self):
return f"<JobExecution(id={self.id}, job_id={self.job_id}, status='{self.status}')>"
class Backup(Base): class Backup(Base):
__tablename__ = "backups" __tablename__ = "backups"
id = Column(Integer, primary_key=True, index=True) id = Column(Integer, primary_key=True)
execution_id = Column(Integer, ForeignKey("job_executions.id"), nullable=False) execution_id = Column(
Integer, ForeignKey("job_executions.id"), nullable=False, index=True
)
storage_path = Column(String, nullable=False) storage_path = Column(String, nullable=False)
size_bytes = Column(Integer, default=0) size_bytes = Column(Integer, default=0)
checksum = Column(String, nullable=True) checksum = Column(String, nullable=True)
type = Column(String, nullable=False) # full, incremental type = Column(String, nullable=False) # full, incremental
parent_backup_id = Column(Integer, ForeignKey("backups.id"), nullable=True) parent_backup_id = Column(
created_at = Column(DateTime, default=datetime.utcnow) Integer, ForeignKey("backups.id"), nullable=True, index=True
)
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
execution = relationship("JobExecution", back_populates="backups") execution = relationship("JobExecution", back_populates="backups")
parent_backup = relationship("Backup", remote_side=[id]) parent_backup = relationship("Backup", remote_side=[id])
def __repr__(self):
return f"<Backup(id={self.id}, execution_id={self.execution_id}, type='{self.type}')>"
class Setting(Base): class Setting(Base):
__tablename__ = "settings" __tablename__ = "settings"
key = Column(String, primary_key=True) key = Column(String, primary_key=True)
value = Column(Text, nullable=True) value = Column(Text, nullable=True)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) updated_at = Column(
DateTime,
default=lambda: datetime.now(timezone.utc),
onupdate=lambda: datetime.now(timezone.utc),
)
def __repr__(self):
return f"<Setting(key='{self.key}', value='{self.value}')>"