best practice of Psycopg

@app.post("/update_error_collection") async def update_error_collection(request: ErrorUpdateRequest): conn = get_db_connection() cur = conn.cursor() try: cur.execute( "INSERT INTO error_collection (input, output, user_id) VALUES (%s, %s, %s)", (request.input_text, request.output_text, request.user_id) ) conn.commit() return {"message": "Error record added successfully"} except Exception as e: conn.rollback() raise HTTPException(status_code=500, detail=f"Error updating error collection: {str(e)}") finally: cur.close() conn.close() def get_db_connection(): return psycopg2.connect(os.environ['DATABASE_URL']) -------------------- Is this best practice? To individually obtain a connection and cursor each time and then close them.

5 Replies

rare-sapphire•13mo ago

In FastAPI you'll want to pass in a reference to a connection pool (rather than just a singel connection) using the Depends() injector, and manage connection lifetimes and transactions using context managers. Typically, you'll have three methods in your database.py : 1. To open up the connection pool when your app starts 2. Pass references to the connection pool 3. Close the connection pool on app shutdown The 2nd one there is what you have with get_db_connection(), and should look something like this

async def get_postgres() -> asyncpg.Pool:
    global conn_pool
    if conn_pool is None:
        logger.error("Connection pool is not initialized.")
        raise ConnectionError("PostgreSQL connection pool is not initialized.")
    try:
        return conn_pool
    except Exception as e:
        logger.error(f"Failed to return PostgreSQL connection pool: {e}")
        raise

async def get_postgres() -> asyncpg.Pool:
    global conn_pool
    if conn_pool is None:
        logger.error("Connection pool is not initialized.")
        raise ConnectionError("PostgreSQL connection pool is not initialized.")
    try:
        return conn_pool
    except Exception as e:
        logger.error(f"Failed to return PostgreSQL connection pool: {e}")
        raise

To actually open up that connection on start, something like this

async def init_postgres() -> None:
    global conn_pool
    try:
        logger.info("Initializing PostgreSQL connection pool...")
        conn_pool = await asyncpg.create_pool(
            dsn=os.getenv("DATABASE_URL"), min_size=1, max_size=10
        )
        logger.info("PostgreSQL connection pool created successfully.")
    except Exception as e:
        logger.error(f"Error initializing PostgreSQL connection pool: {e}")
        raise

async def init_postgres() -> None:
    global conn_pool
    try:
        logger.info("Initializing PostgreSQL connection pool...")
        conn_pool = await asyncpg.create_pool(
            dsn=os.getenv("DATABASE_URL"), min_size=1, max_size=10
        )
        logger.info("PostgreSQL connection pool created successfully.")
    except Exception as e:
        logger.error(f"Error initializing PostgreSQL connection pool: {e}")
        raise

rare-sapphire•13mo ago

Then, when you want to use it in your app, import get_postgres and inject it into your endpoints, like this

@app.post("/update_error_collection", response_model=ErrorUpdateResponse)
async def update_error_collection(
    request: ErrorUpdateRequest,
    db_pool: asyncpg.Pool = Depends(get_postgres),
) -> ErrorUpdateResponse:

@app.post("/update_error_collection", response_model=ErrorUpdateResponse)
async def update_error_collection(
    request: ErrorUpdateRequest,
    db_pool: asyncpg.Pool = Depends(get_postgres),
) -> ErrorUpdateResponse:

To get a connection from the pool and run a query against it, you can use the context manager with db_pool.acquire() such that the connection get automatically when you're done with it

async with db_pool.acquire() as conn:
     results = await conn.fetch(query)

async with db_pool.acquire() as conn:
     results = await conn.fetch(query)

The examples I gave you there are using asyncpg rather than psycopg since it'll use the async functionality of FastAPI. Most of the benefit of async comes from IO bounds operations, like a database query. Let me know if you have other questions. You might find this guide here helpful too!

Neon

Building an Async Product Management API with FastAPI, Pydantic, an...

Learn how to create an asynchronous API for managing products using FastAPI, Pydantic for data validation, and PostgreSQL with connection pooling

jolly-crimsonOP•13mo ago

Thank you very much!! These are very helpful. I have a few more questions: If I don't use await within the route methods in FastAPI, can it cause the entire application to hang? Is the cursor unnecessary in asyncpg ? I using Replit, does using the asyncpg interfere with its integration with Replit?

rare-sapphire•13mo ago

No worries, I'm glad it's useful! In Python, when you don't await an async function call it will actually just return the coroutine object instead of running it which isnt what you want. So it won't hang the app, it'll most likely just crash it. Cursors in asyncpg are not quite the same as in psycopg2, but are still useful in some cases. You'll use a cursor in when you want to iterate over the result of a query without fetching all rows at once. There's a few ways to interact with them, you can see in the asyncpg docs which don't look as nice as the FastAPI onces, but they are really quite good. For Replit, they use uv as the package manager for Python (which is great, uv is absolutely amazing), so you can add it to your project using uv add asyncpg and it should not interfere with anything.

jolly-crimsonOP•13mo ago

Thank you 😆

Gaming

Programming

best practice of Psycopg

Did you find this page helpful?