from datetime import datetime from typing import Optional from sqlalchemy import ( Column, String, Text, Boolean, Integer, BigInteger, DateTime, ForeignKey, CheckConstraint, Index, JSON ) from sqlalchemy.dialects.postgresql import UUID from sqlalchemy.orm import relationship, declarative_base import uuid Base = declarative_base() class Project(Base): __tablename__ = "projects" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) name = Column(String(255), unique=True, nullable=False) description = Column(Text) is_public = Column(Boolean, default=True) created_at = Column(DateTime(timezone=True), default=datetime.utcnow) updated_at = Column(DateTime(timezone=True), default=datetime.utcnow, onupdate=datetime.utcnow) created_by = Column(String(255), nullable=False) packages = relationship("Package", back_populates="project", cascade="all, delete-orphan") permissions = relationship("AccessPermission", back_populates="project", cascade="all, delete-orphan") __table_args__ = ( Index("idx_projects_name", "name"), Index("idx_projects_created_by", "created_by"), ) class Package(Base): __tablename__ = "packages" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) project_id = Column(UUID(as_uuid=True), ForeignKey("projects.id", ondelete="CASCADE"), nullable=False) name = Column(String(255), nullable=False) description = Column(Text) format = Column(String(50), default="generic", nullable=False) platform = Column(String(50), default="any", nullable=False) created_at = Column(DateTime(timezone=True), default=datetime.utcnow) updated_at = Column(DateTime(timezone=True), default=datetime.utcnow, onupdate=datetime.utcnow) project = relationship("Project", back_populates="packages") tags = relationship("Tag", back_populates="package", cascade="all, delete-orphan") uploads = relationship("Upload", back_populates="package", cascade="all, delete-orphan") consumers = relationship("Consumer", back_populates="package", cascade="all, delete-orphan") __table_args__ = ( Index("idx_packages_project_id", "project_id"), Index("idx_packages_name", "name"), Index("idx_packages_format", "format"), Index("idx_packages_platform", "platform"), Index("idx_packages_project_name", "project_id", "name", unique=True), # Composite unique index CheckConstraint( "format IN ('generic', 'npm', 'pypi', 'docker', 'deb', 'rpm', 'maven', 'nuget', 'helm')", name="check_package_format" ), CheckConstraint( "platform IN ('any', 'linux', 'darwin', 'windows', 'linux-amd64', 'linux-arm64', 'darwin-amd64', 'darwin-arm64', 'windows-amd64')", name="check_package_platform" ), {"extend_existing": True}, ) class Artifact(Base): __tablename__ = "artifacts" id = Column(String(64), primary_key=True) # SHA256 hash size = Column(BigInteger, nullable=False) content_type = Column(String(255)) original_name = Column(String(1024)) checksum_md5 = Column(String(32)) # MD5 hash for additional verification checksum_sha1 = Column(String(40)) # SHA1 hash for compatibility s3_etag = Column(String(64)) # S3 ETag for verification artifact_metadata = Column("metadata", JSON, default=dict) # Format-specific metadata (column name is 'metadata') created_at = Column(DateTime(timezone=True), default=datetime.utcnow) created_by = Column(String(255), nullable=False) ref_count = Column(Integer, default=1) s3_key = Column(String(1024), nullable=False) tags = relationship("Tag", back_populates="artifact") uploads = relationship("Upload", back_populates="artifact") @property def sha256(self) -> str: """Alias for id - the SHA256 hash of the artifact content""" return self.id @property def format_metadata(self): """Alias for artifact_metadata - backward compatibility""" return self.artifact_metadata @format_metadata.setter def format_metadata(self, value): """Alias setter for artifact_metadata - backward compatibility""" self.artifact_metadata = value __table_args__ = ( Index("idx_artifacts_created_at", "created_at"), Index("idx_artifacts_created_by", "created_by"), Index("idx_artifacts_ref_count", "ref_count"), # For cleanup queries CheckConstraint("ref_count >= 0", name="check_ref_count_non_negative"), CheckConstraint("size > 0", name="check_size_positive"), ) class Tag(Base): __tablename__ = "tags" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) package_id = Column(UUID(as_uuid=True), ForeignKey("packages.id", ondelete="CASCADE"), nullable=False) name = Column(String(255), nullable=False) artifact_id = Column(String(64), ForeignKey("artifacts.id"), nullable=False) created_at = Column(DateTime(timezone=True), default=datetime.utcnow) updated_at = Column(DateTime(timezone=True), default=datetime.utcnow, onupdate=datetime.utcnow) created_by = Column(String(255), nullable=False) package = relationship("Package", back_populates="tags") artifact = relationship("Artifact", back_populates="tags") history = relationship("TagHistory", back_populates="tag", cascade="all, delete-orphan") __table_args__ = ( Index("idx_tags_package_id", "package_id"), Index("idx_tags_artifact_id", "artifact_id"), Index("idx_tags_package_name", "package_id", "name", unique=True), # Composite unique index Index("idx_tags_package_created_at", "package_id", "created_at"), # For recent tags queries ) class TagHistory(Base): __tablename__ = "tag_history" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) tag_id = Column(UUID(as_uuid=True), ForeignKey("tags.id", ondelete="CASCADE"), nullable=False) old_artifact_id = Column(String(64), ForeignKey("artifacts.id")) new_artifact_id = Column(String(64), ForeignKey("artifacts.id"), nullable=False) change_type = Column(String(20), nullable=False, default="update") changed_at = Column(DateTime(timezone=True), default=datetime.utcnow) changed_by = Column(String(255), nullable=False) tag = relationship("Tag", back_populates="history") __table_args__ = ( Index("idx_tag_history_tag_id", "tag_id"), Index("idx_tag_history_changed_at", "changed_at"), CheckConstraint("change_type IN ('create', 'update', 'delete')", name="check_change_type"), ) class Upload(Base): __tablename__ = "uploads" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) artifact_id = Column(String(64), ForeignKey("artifacts.id"), nullable=False) package_id = Column(UUID(as_uuid=True), ForeignKey("packages.id"), nullable=False) original_name = Column(String(1024)) tag_name = Column(String(255)) # Tag assigned during upload user_agent = Column(String(512)) # Client identification duration_ms = Column(Integer) # Upload timing in milliseconds deduplicated = Column(Boolean, default=False) # Whether artifact was deduplicated checksum_verified = Column(Boolean, default=True) # Whether checksum was verified uploaded_at = Column(DateTime(timezone=True), default=datetime.utcnow) uploaded_by = Column(String(255), nullable=False) source_ip = Column(String(45)) artifact = relationship("Artifact", back_populates="uploads") package = relationship("Package", back_populates="uploads") __table_args__ = ( Index("idx_uploads_artifact_id", "artifact_id"), Index("idx_uploads_package_id", "package_id"), Index("idx_uploads_uploaded_at", "uploaded_at"), Index("idx_uploads_package_uploaded_at", "package_id", "uploaded_at"), Index("idx_uploads_uploaded_by_at", "uploaded_by", "uploaded_at"), ) class Consumer(Base): __tablename__ = "consumers" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) package_id = Column(UUID(as_uuid=True), ForeignKey("packages.id", ondelete="CASCADE"), nullable=False) project_url = Column(String(2048), nullable=False) last_access = Column(DateTime(timezone=True), default=datetime.utcnow) created_at = Column(DateTime(timezone=True), default=datetime.utcnow) package = relationship("Package", back_populates="consumers") __table_args__ = ( Index("idx_consumers_package_id", "package_id"), Index("idx_consumers_last_access", "last_access"), ) class AccessPermission(Base): __tablename__ = "access_permissions" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) project_id = Column(UUID(as_uuid=True), ForeignKey("projects.id", ondelete="CASCADE"), nullable=False) user_id = Column(String(255), nullable=False) level = Column(String(20), nullable=False) created_at = Column(DateTime(timezone=True), default=datetime.utcnow) expires_at = Column(DateTime(timezone=True)) project = relationship("Project", back_populates="permissions") __table_args__ = ( CheckConstraint("level IN ('read', 'write', 'admin')", name="check_level"), Index("idx_access_permissions_project_id", "project_id"), Index("idx_access_permissions_user_id", "user_id"), ) class APIKey(Base): __tablename__ = "api_keys" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) key_hash = Column(String(64), unique=True, nullable=False) name = Column(String(255), nullable=False) user_id = Column(String(255), nullable=False) created_at = Column(DateTime(timezone=True), default=datetime.utcnow) expires_at = Column(DateTime(timezone=True)) last_used = Column(DateTime(timezone=True)) __table_args__ = ( Index("idx_api_keys_user_id", "user_id"), Index("idx_api_keys_key_hash", "key_hash"), ) class AuditLog(Base): __tablename__ = "audit_logs" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) action = Column(String(100), nullable=False) resource = Column(String(1024), nullable=False) user_id = Column(String(255), nullable=False) details = Column(JSON) timestamp = Column(DateTime(timezone=True), default=datetime.utcnow) source_ip = Column(String(45)) __table_args__ = ( Index("idx_audit_logs_action", "action"), Index("idx_audit_logs_resource", "resource"), Index("idx_audit_logs_user_id", "user_id"), Index("idx_audit_logs_timestamp", "timestamp"), Index("idx_audit_logs_resource_timestamp", "resource", "timestamp"), Index("idx_audit_logs_user_timestamp", "user_id", "timestamp"), )