import datetime
from sqlalchemy.orm import declarative_base, relationship
from sqlalchemy.schema import Column, ForeignKey
from sqlalchemy.types import (
BigInteger,
Boolean,
DateTime,
Integer,
Text,
TypeDecorator,
TypeEngine,
)
# Timestamps are saved in the database in UTC without timezone info, so attach
# a UTC timezone to the datetime object
class DateTimeUTC(TypeDecorator):
impl = TypeEngine
cache_ok = True
def load_dialect_impl(self, dialect):
if dialect.name == 'sqlite':
return dialect.type_descriptor(Integer)
return dialect.type_descriptor(DateTime)
def process_bind_param(self, value, dialect):
if dialect.name == 'sqlite':
return value.timestamp() * 1000
return value
def process_result_value(self, value, dialect):
if value is None:
return value
if dialect.name == 'sqlite':
return datetime.datetime.fromtimestamp(value / 1000, datetime.timezone.utc)
return value.replace(tzinfo=datetime.timezone.utc)
Base = declarative_base()
# Note: We have commented out unused columns to keep SQLAlchemy from selecting
# ones that we never end up using
class Backlog(Base):
__tablename__ = 'backlog'
messageid = Column(BigInteger, primary_key=True)
time = Column(DateTimeUTC)
bufferid = Column(Integer, ForeignKey('buffer.bufferid'))
type = Column(Integer)
flags = Column(Integer)
senderid = Column(BigInteger, ForeignKey('sender.senderid'))
senderprefixes = Column(Text)
message = Column(Text)
buffer = relationship('Buffer')
sender = relationship('Sender')
class Sender(Base):
__tablename__ = 'sender'
senderid = Column(BigInteger, primary_key=True)
sender = Column(Text)
# realname = Column(Text)
# avatarurl = Column(Text)
class Buffer(Base):
__tablename__ = 'buffer'
bufferid = Column(Integer, primary_key=True)
userid = Column(Integer, ForeignKey('user.userid'))
groupid = Column(Integer)
networkid = Column(Integer, ForeignKey('network.networkid'))
buffername = Column(Text)
buffercname = Column(Text)
buffertype = Column(Integer)
# lastmsgid = Column(BigInteger)
# lastseenmsgid = Column(BigInteger)
# markerlinemsgid = Column(BigInteger)
# bufferactivity = Column(Integer)
# highlightcount = Column(Integer)
# key = Column(Text)
joined = Column(Boolean)
# cipher = Column(Text)
network = relationship('Network')
class QuasselUser(Base):
__tablename__ = 'quasseluser'
userid = Column(Integer, primary_key=True)
username = Column(Text)
# password = Column(Text)
# hashversion = Column(Integer)
# authenticator = Column(Text)
# The following table is huge and largely irrelevant for quarg, so instead of
# commenting out unused columns, don't even bother listing them all
class Network(Base):
__tablename__ = 'network'
networkid = Column(Integer, primary_key=True)
userid = Column(Integer, ForeignKey('quasseluser.userid'))
networkname = Column(Text)
user = relationship('QuasselUser')
