blob: 77f092a8dd1b857acfaa0f3010ef3f9ae6af1dad (
plain)
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
|
# UGE / L2 / Intro to relational databases / Python project prototype
# Author: Pacien TRAN-GIRARD
# Licence: EUPL-1.2
from os import environ
from functools import partial
from contextlib import contextmanager
import psycopg2
from psycopg2.extras import NamedTupleCursor
from psycopg2.extensions import ISOLATION_LEVEL_READ_COMMITTED
import embrace
from embrace import pool
# Connection to the PostgreSQL database server.
# Using a cursor which exposes result rows as named tuples for convenience.
# TODO: switch to psycopg3 and use asynchronous queries in handlers.
db_connect = partial(
psycopg2.connect,
environ['DATABASE_URL'],
cursor_factory=NamedTupleCursor)
db_pool = pool.ConnectionPool(db_connect, limit=4)
# Turn our annotated SQL queries into Python functions.
queries = embrace.module('./sql/', auto_reload=True)
@contextmanager
def db_transaction(isolation_level=ISOLATION_LEVEL_READ_COMMITTED):
"""
Get a connection from the connection pool and begin a transaction with the
given isolation level. The transaction is automatically rolled back if an
uncaught exception escapes the current context. Otherwise, it is
automatically committed when finishing with no error.
"""
with db_pool.connect() as conn:
conn.set_isolation_level(isolation_level)
with queries.transaction(conn) as tx:
yield tx
|