blob: d9d521eb41bd95e2483cb41dd9a5eff496144fdd (
about) (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
"""module contains all db related stuff"""
from typing import Any, Callable
import MySQLdb as mdb
from redis import Redis
from flask import current_app as app
from gn_libs.mysqldb import database_connection
def with_db_connection(func: Callable[[mdb.Connection], Any]) -> Any:
"""Call `func` with a MySQDdb database connection."""
with database_connection(app.config["SQL_URI"]) as conn:
return func(conn)
def with_redis_connection(func: Callable[[Redis], Any]) -> Any:
"""Call `func` with a redis connection."""
redisuri = app.config["REDIS_URL"]
with Redis.from_url(redisuri, decode_responses=True) as rconn:
return func(rconn)
|
