import datetime from sqlalchemy.schema import Column, ForeignKey from sqlalchemy.types import BigInteger, Boolean, DateTime, Integer, Text, TypeDecorator, TypeEngine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship # pylint: disable=too-few-public-methods # Timestamps are saved in the database in UTC without timezone info, so attach # a UTC timezone to the datetime object class DateTimeUTC(TypeDecorator): # pylint complains that process_literal_param and python_type are # abstract but not overriden. This seems to not be necessary with # SQLAlchemy, so squash those warnings # pylint: disable=abstract-method 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')