-
Notifications
You must be signed in to change notification settings - Fork 157
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Alexey Firsov
committed
Nov 29, 2017
1 parent
fea8aa5
commit d825cb6
Showing
9 changed files
with
632 additions
and
16 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,191 @@ | ||
| import asyncio | ||
| import enum | ||
| import uuid | ||
| import warnings | ||
| from abc import ABC, abstractmethod | ||
|
|
||
| import psycopg2 | ||
| from aiopg.utils import PY_35, _TransactionPointContextManager | ||
|
|
||
| __all__ = ('IsolationLevel', 'Transaction') | ||
|
|
||
|
|
||
| class IsolationCompiler(ABC): | ||
| name = '' | ||
|
|
||
| __slots__ = ('_readonly', '_deferrable') | ||
|
|
||
| def __init__(self, readonly, deferrable): | ||
| self._readonly = readonly | ||
| self._deferrable = deferrable | ||
| self._check_readonly_deferrable() | ||
|
|
||
| def _check_readonly_deferrable(self): | ||
| available = self._readonly or self._deferrable | ||
| if not isinstance(self, SerializableCompiler) and available: | ||
| raise ValueError('Is only available for serializable transactions') | ||
|
|
||
| def savepoint(self, unique_id): | ||
| return 'SAVEPOINT {}'.format(unique_id) | ||
|
|
||
| def release_savepoint(self, unique_id): | ||
| return 'RELEASE SAVEPOINT {}'.format(unique_id) | ||
|
|
||
| def rollback_savepoint(self, unique_id): | ||
| return 'ROLLBACK TO SAVEPOINT {}'.format(unique_id) | ||
|
|
||
| def commit(self): | ||
| return 'COMMIT' | ||
|
|
||
| def rollback(self): | ||
| return 'ROLLBACK' | ||
|
|
||
| @abstractmethod | ||
| def begin(self): | ||
| raise NotImplementedError("Please Implement this method") | ||
|
|
||
| def __repr__(self): | ||
| return self.name | ||
|
|
||
|
|
||
| class ReadCommittedCompiler(IsolationCompiler): | ||
| name = 'Read committed' | ||
|
|
||
| def begin(self): | ||
| return 'BEGIN' | ||
|
|
||
|
|
||
| class RepeatableReadCompiler(IsolationCompiler): | ||
| name = 'Repeatable read' | ||
|
|
||
| def begin(self): | ||
| return 'BEGIN ISOLATION LEVEL REPEATABLE READ' | ||
|
|
||
|
|
||
| class SerializableCompiler(IsolationCompiler): | ||
| name = 'Serializable' | ||
|
|
||
| def begin(self): | ||
| query = 'BEGIN ISOLATION LEVEL SERIALIZABLE' | ||
|
|
||
| if self._readonly: | ||
| query += ' READ ONLY' | ||
|
|
||
| if self._deferrable: | ||
| query += ' DEFERRABLE' | ||
|
|
||
| return query | ||
|
|
||
|
|
||
| class IsolationLevel(enum.Enum): | ||
| serializable = SerializableCompiler | ||
| repeatable_read = RepeatableReadCompiler | ||
| read_committed = ReadCommittedCompiler | ||
|
|
||
| def __call__(self, readonly, deferrable): | ||
| return self.value(readonly, deferrable) | ||
|
|
||
|
|
||
| class Transaction: | ||
| __slots__ = ('_cur', '_is_begin', '_isolation', '_unique_id') | ||
|
|
||
| def __init__(self, cur, isolation_level, | ||
| readonly=False, deferrable=False): | ||
| self._cur = cur | ||
| self._is_begin = False | ||
| self._unique_id = None | ||
| self._isolation = isolation_level(readonly, deferrable) | ||
|
|
||
| @property | ||
| def is_begin(self): | ||
| return self._is_begin | ||
|
|
||
| @asyncio.coroutine | ||
| def begin(self): | ||
| if self._is_begin: | ||
| raise psycopg2.ProgrammingError( | ||
| 'You are trying to open a new transaction, use the save point') | ||
| self._is_begin = True | ||
| yield from self._cur.execute(self._isolation.begin()) | ||
| return self | ||
|
|
||
| @asyncio.coroutine | ||
| def commit(self): | ||
| self._check_commit_rollback() | ||
| yield from self._cur.execute(self._isolation.commit()) | ||
| self._is_begin = False | ||
|
|
||
| @asyncio.coroutine | ||
| def rollback(self): | ||
| self._check_commit_rollback() | ||
| yield from self._cur.execute(self._isolation.rollback()) | ||
| self._is_begin = False | ||
|
|
||
| @asyncio.coroutine | ||
| def rollback_savepoint(self): | ||
| self._check_release_rollback() | ||
| yield from self._cur.execute( | ||
| self._isolation.rollback_savepoint(self._unique_id)) | ||
| self._unique_id = None | ||
|
|
||
| @asyncio.coroutine | ||
| def release_savepoint(self): | ||
| self._check_release_rollback() | ||
| yield from self._cur.execute( | ||
| self._isolation.release_savepoint(self._unique_id)) | ||
| self._unique_id = None | ||
|
|
||
| @asyncio.coroutine | ||
| def savepoint(self): | ||
| self._check_commit_rollback() | ||
| if self._unique_id is not None: | ||
| raise psycopg2.ProgrammingError('You do not shut down savepoint') | ||
|
|
||
| self._unique_id = 's{}'.format(uuid.uuid1().hex) | ||
| yield from self._cur.execute( | ||
| self._isolation.savepoint(self._unique_id)) | ||
|
|
||
| return self | ||
|
|
||
| def point(self): | ||
| return _TransactionPointContextManager(self.savepoint()) | ||
|
|
||
| def _check_commit_rollback(self): | ||
| if not self._is_begin: | ||
| raise psycopg2.ProgrammingError('You are trying to commit ' | ||
| 'the transaction does not open') | ||
|
|
||
| def _check_release_rollback(self): | ||
| self._check_commit_rollback() | ||
| if self._unique_id is None: | ||
| raise psycopg2.ProgrammingError('You do not start savepoint') | ||
|
|
||
| def __repr__(self): | ||
| return "<{} transaction={} id={:#x}>".format( | ||
| self.__class__.__name__, | ||
| self._isolation, | ||
| id(self) | ||
| ) | ||
|
|
||
| def __del__(self): | ||
| if self._is_begin: | ||
| warnings.warn( | ||
| "You have not closed transaction {!r}".format(self), | ||
| ResourceWarning) | ||
|
|
||
| if self._unique_id is not None: | ||
| warnings.warn( | ||
| "You have not closed savepoint {!r}".format(self), | ||
| ResourceWarning) | ||
|
|
||
| if PY_35: | ||
| @asyncio.coroutine | ||
| def __aenter__(self): | ||
| return (yield from self.begin()) | ||
|
|
||
| @asyncio.coroutine | ||
| def __aexit__(self, exc_type, exc, tb): | ||
| if exc_type is not None: | ||
| yield from self.rollback() | ||
| else: | ||
| yield from self.commit() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| import asyncio | ||
|
|
||
| import aiopg | ||
| from aiopg.transaction import Transaction, IsolationLevel | ||
|
|
||
| dsn = 'dbname=aiopg user=aiopg password=passwd host=127.0.0.1' | ||
|
|
||
|
|
||
| async def transaction(cur, isolation_level, | ||
| readonly=False, deferrable=False): | ||
| async with Transaction(cur, isolation_level, | ||
| readonly, deferrable) as transaction: | ||
| await cur.execute('insert into tbl values (1)') | ||
|
|
||
| async with transaction.point(): | ||
| await cur.execute('insert into tbl values (3)') | ||
|
|
||
| await cur.execute('insert into tbl values (4)') | ||
|
|
||
|
|
||
| async def main(): | ||
| async with aiopg.create_pool(dsn) as pool: | ||
| async with pool.acquire() as conn: | ||
| async with conn.cursor() as cur: | ||
| await cur.execute('CREATE TABLE tbl (id int)') | ||
| await transaction(cur, IsolationLevel.repeatable_read) | ||
| await transaction(cur, IsolationLevel.read_committed) | ||
| await transaction(cur, IsolationLevel.serializable) | ||
|
|
||
| await cur.execute('select * from tbl') | ||
|
|
||
|
|
||
| loop = asyncio.get_event_loop() | ||
| loop.run_until_complete(main()) |
Oops, something went wrong.