Last active
January 11, 2023 13:52
-
-
Save vulcan25/55ce270d76bf78044d067c51e23ae5ad to your computer and use it in GitHub Desktop.
psycopg2 flask implementation with connection pooling support
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
from flask import Flask, g, jsonify | |
import werkzeug, os | |
from werkzeug.utils import secure_filename | |
import psycopg2 | |
from psycopg2 import pool | |
def get_db(): | |
print ('GETTING CONN') | |
if 'db' not in g: | |
g.db = app.config['postgreSQL_pool'].getconn() | |
return g.db | |
def create_app(): | |
app = Flask(__name__) | |
app.config['postgreSQL_pool'] = psycopg2.pool.SimpleConnectionPool(1, 20,user = "postgres", | |
password = "top_secret", | |
host = "127.0.0.1", | |
port = "9502", | |
database = "postgres") | |
@app.teardown_appcontext | |
def close_conn(e): | |
print('CLOSING CONN') | |
db = g.pop('db', None) | |
if db is not None: | |
app.config['postgreSQL_pool'].putconn(db) | |
@app.route('/') | |
def index(): | |
print ('ROUTE') | |
db = get_db() | |
cursor = db.cursor() | |
cursor.execute("select 1;") | |
result = cursor.fetchall() | |
print (result) | |
cursor.close() | |
return jsonify(result) | |
return app | |
if __name__ == '__main__': | |
app = create_app() | |
app.run(debug=True) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Thanks for simple, but useful example. Just want to emphasize
Note: This pool class is useful only for single-threaded applications.
From psycopg2 documentation. So, it better to use ThreadedConnectionPool, i think.