-
Notifications
You must be signed in to change notification settings - Fork 12
/
orm.py
151 lines (119 loc) · 4.48 KB
/
orm.py
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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
import collections
import logging
import typing
import uuid
import sqlalchemy
from sqlalchemy import (Table, Column, MetaData, String, Integer, Text,
ForeignKey, create_engine, event)
from sqlalchemy.orm import mapper, scoped_session, sessionmaker, composite, relationship
import sqlalchemy.exc
import sqlalchemy.orm.exc
from sqlalchemy_utils.functions import create_database, drop_database
from sqlalchemy_utils.types.uuid import UUIDType
from issues.domain.model import Issue, IssueReporter, Assignment
from issues.domain.ports import (IssueLog, UnitOfWork, UnitOfWorkManager,
MessageBus)
SessionFactory = typing.Callable[[], sqlalchemy.orm.Session]
class IssueRepository(IssueLog):
def __init__(self, session):
self._session = session
def add(self, issue: Issue) -> None:
self._session.add(issue)
def _get(self, issue_id) -> Issue:
return self._session.query(Issue).\
filter_by(id=issue_id).\
first()
class SqlAlchemyUnitOfWork(UnitOfWork):
def __init__(self, sessionfactory: SessionFactory, bus: MessageBus) -> None:
self.sessionfactory = sessionfactory
self.bus = bus
event.listen(self.sessionfactory, "after_flush", self.gather_events)
event.listen(self.sessionfactory, "loaded_as_persistent",
self.setup_events)
def __enter__(self):
self.session = self.sessionfactory()
self.flushed_events = []
return self
def __exit__(self, type, value, traceback):
self.publish_events()
def commit(self):
self.session.flush()
self.session.commit()
def rollback(self):
self.flushed_events = []
self.session.rollback()
def setup_events(self, session, entity):
entity.events = []
def gather_events(self, session, ctx):
flushed_objects = [e for e in session.new] + [e for e in session.dirty]
for e in flushed_objects:
try:
self.flushed_events += e.events
except AttributeError:
pass
def publish_events(self):
for e in self.flushed_events:
self.bus.handle(e)
@property
def issues(self):
return IssueRepository(self.session)
class SqlAlchemy:
def __init__(self, uri, bus):
self.engine = create_engine(uri)
self.bus = bus
self._session_maker = scoped_session(sessionmaker(self.engine),)
def recreate_schema(self):
self.configure_mappings()
drop_database(self.engine.url)
self.create_schema()
def get_session(self):
return self._session_maker()
def start_unit_of_work(self):
return SqlAlchemyUnitOfWork(self._session_maker, self.bus)
def create_schema(self):
create_database(self.engine.url)
self.metadata.create_all()
def configure_mappings(self):
self.metadata = MetaData(self.engine)
IssueReporter.__composite_values__ = lambda i: (i.name, i.email)
issues = Table('issues', self.metadata,
Column('pk', Integer, primary_key=True),
Column('issue_id', UUIDType),
Column('reporter_name', String(50)),
Column('reporter_email', String(50)),
Column('description', Text))
assignments = Table(
'assignments',
self.metadata,
Column('pk', Integer, primary_key=True),
Column('id', UUIDType),
Column('fk_assignment_id', UUIDType, ForeignKey('issues.issue_id')),
Column('assigned_by', String(50)),
Column('assigned_to', String(50)),
)
mapper(
Issue,
issues,
properties={
'__pk':
issues.c.pk,
'id':
issues.c.issue_id,
'description':
issues.c.description,
'reporter':
composite(IssueReporter, issues.c.reporter_name,
issues.c.reporter_email),
'_assignments':
relationship(Assignment, backref='issue')
},
),
mapper(
Assignment,
assignments,
properties={
'__pk': assignments.c.pk,
'id': assignments.c.id,
'assigned_to': assignments.c.assigned_to,
'assigned_by': assignments.c.assigned_by
})