Skip to content

Instantly share code, notes, and snippets.

@vulcan25
Last active January 11, 2023 13:52
Show Gist options
  • Star 14 You must be signed in to star a gist
  • Fork 4 You must be signed in to fork a gist
  • Save vulcan25/55ce270d76bf78044d067c51e23ae5ad to your computer and use it in GitHub Desktop.
Save vulcan25/55ce270d76bf78044d067c51e23ae5ad to your computer and use it in GitHub Desktop.
psycopg2 flask implementation with connection pooling support
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)
@2uger
Copy link

2uger commented Oct 10, 2021

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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment