#!/usr/bin/env python3 import enum from typing import Any from typing import Optional from typing import Union from app import activitypub as ap from app.database import Base from app.database import metadata_obj from app.activitypub import BaseActor from sqlalchemy import Column from sqlalchemy import Boolean from sqlalchemy import DateTime from sqlalchemy import Enum from sqlalchemy import Integer from sqlalchemy import String from sqlalchemy import JSON from sqlalchemy import ForeignKey from sqlalchemy.orm import Mapped from sqlalchemy.orm import relationship from datetime import datetime as dtime from datetime import timezone def now() -> dtime: return dtime.now(timezone.utc) class Actor(Base, BaseActor): __tablename__ = "actor" id = Column(Integer, primary_key=True, index=True) created_at = Column(DateTime(timezone=True), nullable=False, default=now) updated_at = Column(DateTime(timezone=True), nullable=False, default=now) ap_id: Mapped[str] = Column(String, unique=True, nullable=False, index=True) # type: ignore ap_actor = Column(JSON, nullable=False) ap_type = Column(String, nullable=False) handle = Column(String, nullable=True, index=True) is_blocked = Column(Boolean, nullable=False, default=False, server_default="0") is_deleted = Column(Boolean, nullable=False, default=False, server_default="0") class InboxObject(Base): __tablename__ = "inbox" id = Column(Integer, primary_key=True, index=True) created_at = Column(DateTime(timezone=True), nullable=False, default=now) updated_at = Column(DateTime(timezone=True), nullable=False, default=now) actor_id = Column(Integer, ForeignKey("actor.id"), nullable=False) actor: Mapped[Actor] = relationship(Actor, uselist=False) server = Column(String, nullable=False) is_hidden_from_stream = Column(Boolean, nullable=False, default=False) ap_actor_id = Column(String, nullable=False) ap_type = Column(String, nullable=False, index=True) ap_id: Mapped[str] = Column(String, nullable=False, unique=True, index=True) # type: ignore ap_context = Column(String, nullable=True) ap_published_at = Column(DateTime(timezone=True), nullable=False) ap_object: Mapped[dict[str, Any] | None] = Column(JSON, nullable=False) # type: ignore activity_object_ap_id = Column(String, nullable=True, index=True) visibility = Column(Enum(ap.VisibilityEnum), nullable=False) is_deleted = Column(Boolean, nullable=False, default=False) relates_to_inbox_object_id = Column( Integer, ForeignKey("inbox.id"), nullable=True, ) relates_to_inbox_object: Mapped[Optional["InboxObject"]] = relationship( "InboxObject", foreign_keys=[relates_to_inbox_object_id], remote_side=id, uselist=False, ) relates_to_outbox_object_id = Column( Integer, ForeignKey("outbox.id", use_alter=True), nullable=True, ) relates_to_outbox_object: Mapped[Optional["OutboxObject"]] = relationship( "OutboxObject", foreign_keys=relates_to_outbox_object_id, uselist=False, ) class OutboxObject(Base): __tablename__ = "outbox" id = Column(Integer, primary_key=True, index=True) created_at = Column(DateTime(timezone=True), nullable=False, default=now) updated_at = Column(DateTime(timezone=True), nullable=False, default=now) public_id = Column(String, nullable=False, index=True) slug = Column(String, nullable=True, index=True) ap_type = Column(String, nullable=False, index=True) ap_id: Mapped[str] = Column(String, nullable=False, unique=True, index=True) # type: ignore ap_context = Column(String, nullable=True) ap_object: Mapped[ap.RawObject] = Column(JSON, nullable=False) #type: ignore activity_object_ap_id = Column(String, nullable=True, index=True) visibility = Column(Enum(ap.VisibilityEnum), nullable=False) is_deleted = Column(Boolean, nullable=False, default=False) likes_count = Column(Integer, nullable=False, default=0) announces_count = Column(Integer, nullable=False, default=0) replies_count: Mapped[int] = Column(Integer, nullable=False, default=0) #type: ignore relates_to_inbox_object_id = Column( Integer, ForeignKey("inbox.id", use_alter=True), nullable=True, ) relates_to_inbox_object: Mapped[Optional["InboxObject"]] = relationship( "InboxObject", foreign_keys=relates_to_inbox_object_id, uselist=False, ) relates_to_outbox_object_id = Column( Integer, ForeignKey("outbox.id"), nullable=True, ) relates_to_outbox_object: Mapped[Optional["OutboxObject"]] = relationship( "OutboxObject", foreign_keys=[relates_to_outbox_object_id], remote_side=id, uselist=False, ) relates_to_actor_id = Column( Integer, ForeignKey("actor.id"), nullable=True, ) relates_to_actor: Mapped[Optional["Actor"]] = relationship( "Actor", foreign_keys=[relates_to_actor_id], uselist=False, ) class IncomingActivity(Base): __tablename__ = "ingress" id = Column(Integer, primary_key=True, index=True) created_at = Column(DateTime(timezone=True), nullable=False, default=now) ap_id = Column(String, nullable=True, index=True) ap_object = Column(JSON, nullable=True) class OutcomingActivity(Base): __tablename__ = "push" id = Column(Integer, primary_key=True, index=True) created_at = Column(DateTime(timezone=True), nullable=False, default=now) ap_id = Column(String, nullable=True, index=True) ap_object = Column(JSON, nullable=True) class Follower(Base): __tablename__ = "follower" id = Column(Integer, primary_key=True, index=True) created_at = Column(DateTime(timezone=True), nullable=False, default=now) updated_at = Column(DateTime(timezone=True), nullable=False, default=now) actor_id = Column(Integer, ForeignKey("actor.id"), nullable=False, unique=True) actor: Mapped[Actor] = relationship(Actor, uselist=False) inbox_object_id = Column(Integer, ForeignKey("inbox.id"), nullable=False) inbox_object = relationship(InboxObject, uselist=False) ap_actor_id = Column(String, nullable=False, unique=True) class Following(Base): __tablename__ = "following" id = Column(Integer, primary_key=True, index=True) created_at = Column(DateTime(timezone=True), nullable=False, default=now) updated_at = Column(DateTime(timezone=True), nullable=False, default=now) actor_id = Column(Integer, ForeignKey("actor.id"), nullable=False, unique=True) actor: Mapped[Actor] = relationship(Actor, uselist=False) outbox_object_id = Column(Integer, ForeignKey("outbox.id"), nullable=False) outbox_object = relationship(OutboxObject, uselist=False) ap_actor_id = Column(String, nullable=False, unique=True)