1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
# pylint: disable=too-few-public-methods, invalid-name
from sqlalchemy import Column, ForeignKey
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
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)
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)
begin_date = Column(UTCDateTime, index=True)
end_date = Column(UTCDateTime)
all_day = Column(Boolean)
name = Column(String(255))
location = Column(String(255))
_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
__all__ = ('Calendar', 'Entry', 'Category')
|
