#!/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 = Column(String, unique=True, nullable=False, index=True) 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 = Column(String, nullable=False, unique=True, index=True) ap_context = Column(String, nullable=True) ap_published_at = Column(DateTime(timezone=True), nullable=False) ap_object = Column(JSON, nullable=False) 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 = Column(String, nullable=False, unique=True, index=True) ap_context = Column(String, nullable=True) ap_object = Column(JSON, nullable=False) 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 = Column(Integer, nullable=False, default=0) 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)