PostgreSQL API with Python: Build & Connect

Connecting Python to PostgreSQL is a powerful skill. It lets you build dynamic applications. An API acts as a bridge between your database and the outside world. This guide shows you how to create one.

We will use popular Python tools. FastAPI will handle web requests. SQLAlchemy will manage database interactions. This combination is fast and reliable.

Why Use PostgreSQL with Python?

PostgreSQL is a robust, open-source database. It handles complex queries and large datasets well. Python is a versatile programming language. Together, they form a strong backend foundation.

An API allows other software to interact with your database safely. It controls what data can be read or written. This is essential for modern web and mobile apps.

For a broader look at creating interfaces, see our Python API Tutorial for Beginners.

Project Setup and Installation

First, ensure you have Python installed. You will also need a running PostgreSQL database. You can install one locally or use a cloud service.

Create a new project directory. Then, install the required Python packages. Use the following command in your terminal.

pip install fastapi uvicorn sqlalchemy psycopg2-binary pydantic
    

Here is what each package does:

• FastAPI: Web framework for building the API.
• Uvicorn: ASGI server to run the FastAPI app.
• SQLAlchemy: ORM (Object-Relational Mapper) for Python.
• Psycopg2: PostgreSQL adapter for Python.
• Pydantic: Data validation and settings management.

Connecting to the PostgreSQL Database

The first step is to establish a connection. We use SQLAlchemy to create an engine. This engine manages database connections.

Create a file named `database.py`. Add the following code to it.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# Connection string format: postgresql://user:password@host:port/database
DATABASE_URL = "postgresql://myuser:mypassword@localhost:5432/mydatabase"

# Create the SQLAlchemy engine
engine = create_engine(DATABASE_URL)

# Create a configured "SessionLocal" class
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Create a Base class for our models
Base = declarative_base()
    

Replace the connection string with your own details. The create_engine function sets up the connection pool. The SessionLocal class will be used to get a database session.

Defining the Data Model with SQLAlchemy

A model represents a table in your database. Let's create a simple model for a "tasks" application. Create a file named `models.py`.

from sqlalchemy import Column, Integer, String, Boolean
from database import Base

class Task(Base):
    __tablename__ = "tasks"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, nullable=False)
    description = Column(String)
    is_completed = Column(Boolean, default=False)
    

This code defines a Task table. It has four columns: id, title, description, and is_completed. The Base.metadata.create_all(bind=engine) command can create this table in PostgreSQL.

Creating Pydantic Schemas

Schemas define the structure of request and response data. They are separate from the database model. This is a key feature of FastAPI. Create a `schemas.py` file.

from pydantic import BaseModel

class TaskBase(BaseModel):
    title: str
    description: str | None = None

class TaskCreate(TaskBase):
    pass  # Inherits from TaskBase

class Task(TaskBase):
    id: int
    is_completed: bool

    class Config:
        orm_mode = True
    

The TaskCreate schema is for creating a new task. The Task schema includes the id and status, and is used for responses. The orm_mode tells Pydantic to read data from ORM objects.

Building the FastAPI Application

Now, let's build the main API application. We will create endpoints to Create, Read, Update, and Delete (CRUD) tasks. Create a `main.py` file.

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List

import models, schemas
from database import SessionLocal, engine

# Create the database tables
models.Base.metadata.create_all(bind=engine)

app = FastAPI()

# Dependency to get the database session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/tasks/", response_model=schemas.Task)
def create_task(task: schemas.TaskCreate, db: Session = Depends(get_db)):
    db_task = models.Task(**task.dict())
    db.add(db_task)
    db.commit()
    db.refresh(db_task)
    return db_task

@app.get("/tasks/", response_model=List[schemas.Task])
def read_tasks(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    tasks = db.query(models.Task).offset(skip).limit(limit).all()
    return tasks

@app.get("/tasks/{task_id}", response_model=schemas.Task)
def read_task(task_id: int, db: Session = Depends(get_db)):
    task = db.query(models.Task).filter(models.Task.id == task_id).first()
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

@app.put("/tasks/{task_id}", response_model=schemas.Task)
def update_task(task_id: int, task_update: schemas.TaskCreate, db: Session = Depends(get_db)):
    db_task = db.query(models.Task).filter(models.Task.id == task_id).first()
    if db_task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    for key, value in task_update.dict().items():
        setattr(db_task, key, value)
    db.commit()
    db.refresh(db_task)
    return db_task

@app.delete("/tasks/{task_id}")
def delete_task(task_id: int, db: Session = Depends(get_db)):
    db_task = db.query(models.Task).filter(models.Task.id == task_id).first()
    if db_task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    db.delete(db_task)
    db.commit()
    return {"message": "Task deleted successfully"}
    

The get_db function provides a database session for each request. The Depends function is a FastAPI dependency injection. Each endpoint performs a specific CRUD operation.

Running and Testing the API

Start the server with Uvicorn. Run this command in your terminal from the project directory.

uvicorn main:app --reload
    

Open your browser and go to