# pylint: disable=too-few-public-methods, invalid-name from sqlalchemy import Column, ForeignKey, Table from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship from sqlalchemy.types import (TIMESTAMP, Boolean, DateTime, Integer, String, TypeDecorator) from dateutil.tz import tzutc # pylint: disable=abstract-method, unused-argument class UTCDateTime(TypeDecorator): impl = DateTime def process_bind_param(self, value, engine): if value is not None: return value.astimezone(tzutc()) def process_result_value(self, value, engine): if value is not None: return value.replace(tzinfo=tzutc()) BASE = declarative_base() class Calendar(BASE): __tablename__ = 'calendars' uid = Column(String(255), primary_key=True) url = Column(String(255)) name = Column(String(255), index=True) website = Column(String(255)) visible = Column(Boolean, index=True) custom_name = Column(String(255)) custom_image = Column(String(255)) custom_url = Column(String(255)) last_updated = Column(TIMESTAMP) _category = Column( Integer, ForeignKey( 'categories.id', onupdate='CASCADE', ondelete='SET NULL')) category = relationship( 'Category', back_populates='calendars', order_by='Calendar.name') entries = relationship( 'Entry', back_populates='calendar', cascade="all", passive_deletes=True, order_by='Entry.begin_date') @staticmethod def fetch(uid, session, name=None, url=None): calendar = session.query(Calendar).filter(Calendar.uid == uid).first() if not calendar: calendar = Calendar() calendar.uid = uid session.add(calendar) if name: calendar.name = name if url: calendar.url = url return calendar class Entry(BASE): __tablename__ = 'entries' id = Column(Integer, primary_key=True) uid = Column(String(255), index=True, unique=True, nullable=False) begin_date = Column(UTCDateTime, index=True) end_date = Column(UTCDateTime) all_day = Column(Boolean) name = Column(String(255)) location = Column(String(255)) last_modified = Column(UTCDateTime) _calendar = Column( String(255), ForeignKey( 'calendars.uid', onupdate='CASCADE', ondelete='CASCADE')) calendar = relationship( 'Calendar', back_populates='entries', order_by='Entry.begin_date') @staticmethod def fetch(uid, session): entry = session.query(Entry).filter(Entry.uid == uid).first() if not entry: entry = Entry() session.add(entry) return entry class Category(BASE): __tablename__ = 'categories' id = Column(Integer, primary_key=True) name = Column(String(255), index=True) calendars = relationship( 'Calendar', back_populates='category', cascade="all", passive_deletes=True, order_by='Calendar.name') @staticmethod def fetch(name, session): category = session.query(Category).filter( Category.name == name).first() if not category: category = Category() category.name = name session.add(category) return category user_selections = Table( 'user_selections', BASE.metadata, Column('_user', Integer, ForeignKey('users.id', onupdate='CASCADE', ondelete='CASCADE'), primary_key=True), Column('_calendar', String(255), ForeignKey('calendars.uid', onupdate='CASCADE', ondelete='CASCADE'), primary_key=True) ) class User(BASE): __tablename__ = 'users' id = Column(Integer, primary_key=True) login = Column(String(255), unique=True, index=True) password = Column(String(255)) is_admin = Column(Boolean) timezone = Column(String(255)) last_login = Column(UTCDateTime) calendars = relationship('Calendar', secondary=user_selections) __all__ = ('Calendar', 'Entry', 'Category', 'User')