-
Notifications
You must be signed in to change notification settings - Fork 0
/
unit_of_work.py
78 lines (63 loc) · 1.61 KB
/
unit_of_work.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
# -*- coding: utf-8 -*-
#
# This module is part of the Frequent project, Copyright (C) 2019,
# Douglas Daly. The Frequent package is free software, licensed under
# the MIT License.
#
# Source Code:
# https://github.com/douglasdaly/frequent-py
# Documentation:
# https://frequent-py.readthedocs.io/en/latest
# License:
# https://frequent-py.readthedocs.io/en/latest/license.html
#
"""
Unit of Work pattern base classes for creating units of work.
"""
from abc import ABC
from abc import abstractmethod
from typing import Type
class UnitOfWork(ABC):
"""
Base class for units of work.
See Also
--------
UnitOfWorkManager
"""
def __enter__(self) -> 'UnitOfWork':
return self
def __exit__(
self,
exc_type: Type[Exception],
exc_value: Exception,
traceback
) -> None:
if exc_type is not None:
self.rollback()
else:
self.commit()
return
@abstractmethod
def commit(self) -> None:
"""Commits (persists) changes made during this unit of work."""
pass
@abstractmethod
def rollback(self) -> None:
"""Rolls back any changes made during this unit of work."""
pass
class UnitOfWorkManager(ABC):
"""
Manager base class for creating :obj:`UnitOfWork` instances.
See Also
--------
UnitOfWork
"""
@abstractmethod
def start(self) -> UnitOfWork:
"""Creates a new unit of work object to use.
Returns
-------
UnitOfWork
The new unit of work object to use.
"""
pass