| 629 | |
| 630 | |
| 631 | class Storage(Base): |
| 632 | __tablename__: str = "storage" |
| 633 | |
| 634 | id: Mapped[str] = mapped_column( |
| 635 | String(32), primary_key=True, default=lambda: token_hex(16) |
| 636 | ) |
| 637 | name: Mapped[str] = mapped_column(String(100), index=True) |
| 638 | type: Mapped[str] = mapped_column( |
| 639 | SQLAEnum("database", "volume", "kv", "queue", name="storage_type"), |
| 640 | nullable=False, |
| 641 | ) |
| 642 | status: Mapped[str] = mapped_column( |
| 643 | SQLAEnum("pending", "active", "resetting", "deleted", name="storage_status"), |
| 644 | nullable=False, |
| 645 | default="pending", |
| 646 | ) |
| 647 | config: Mapped[dict[str, object]] = mapped_column( |
| 648 | JSONB, nullable=False, default=dict |
| 649 | ) |
| 650 | error: Mapped[dict[str, object] | None] = mapped_column(JSONB, nullable=True) |
| 651 | created_by_user_id: Mapped[int | None] = mapped_column( |
| 652 | ForeignKey("user.id", use_alter=True, ondelete="SET NULL"), nullable=True |
| 653 | ) |
| 654 | created_at: Mapped[datetime] = mapped_column( |
| 655 | index=True, nullable=False, default=utc_now |
| 656 | ) |
| 657 | updated_at: Mapped[datetime] = mapped_column( |
| 658 | index=True, nullable=False, default=utc_now, onupdate=utc_now |
| 659 | ) |
| 660 | team_id: Mapped[str] = mapped_column(ForeignKey("team.id"), index=True) |
| 661 | |
| 662 | # Relationships |
| 663 | team: Mapped["Team"] = relationship(back_populates="storages") |
| 664 | created_by_user: Mapped[User | None] = relationship( |
| 665 | foreign_keys=[created_by_user_id] |
| 666 | ) |
| 667 | project_links: Mapped[list["StorageProject"]] = relationship( |
| 668 | back_populates="storage" |
| 669 | ) |
| 670 | |
| 671 | __table_args__ = ( |
| 672 | UniqueConstraint("team_id", "name", name="uq_storage_team_name"), |
| 673 | Index( |
| 674 | "ix_storage_team_name_lower", |
| 675 | "team_id", |
| 676 | func.lower(name), |
| 677 | unique=True, |
| 678 | ), |
| 679 | ) |
| 680 | |
| 681 | @override |
| 682 | def __repr__(self): |
| 683 | return f"<Storage {self.name} ({self.type})>" |
| 684 | |
| 685 | @property |
| 686 | def projects(self) -> list["Project"]: |
| 687 | return [link.project for link in self.project_links if link.project] |
| 688 |
no outgoing calls
no test coverage detected