SQLAlchemy setup

Installation

• To install run

pip3 install sqlalchemy
pip3 install psycopg

Note: SQLAlchemy has all the code needed for us to interact with the database and write python code to do so, but it does not know how to communicate with the Database itself.

• Having said that, we would need to install the driver separately

• In our Postgres case that would be psycopg

• Usually all ORMs work like this where you need the driver separately

File Structure

.
└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    └── schemas.py

Guidelines

• Once installed we will create a file called database.py in our App folder

• This file will handle our database connection

• In this file we will have the following:

# Imports for SQLAlchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# Database connection string format
SQLALCHEMY_DATABASE_URL = 'postgresql://<username>:<password>@<ip-address/hostname>/<databasename>'

# Example:
# SQLALCHEMY_DATABASE_URL = 'postgresql://postgres:Password1@localhost/fastapi'

# Create engine for interacting with database
engine = create_engine(SQLALCHEMY_DATABASE_URL)

# However if we want to interact with the database
# We will need to set up a session
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Note the sessionmaker params are default values

# We additionally need to define our base class
# This is used for all models that we will be defining
Base = declarative_base()

• Now that this is set up, we will create a new file called models.py

• This is where all our database tables will be created

• Each model is a table

• The file contains the following:

# Base needs to be imported from our database file
from app.database import Base

# Columns and Datatypes required for SQL
from sqlalchemy import Column, Integer, String, Boolean

# Different expressions
from sqlalchemy.sql.expression import null

# Define class
class Post(Base):

    # Define Tablename
    __tablename__ = 'posts'

    # Define Columns
    id = Column(Integer, primary_key=True, nullable=False)
    title = Column(String, nullable=False)
    content = Column(String, nullable=False)
    published = Column(Boolean, server_default='TRUE', nullable=False)
    
# Note: Template for Column is
# Column(DATATYPE?, PRIMARY_KEY?, NULLABLE?, DEFAULT VALUE?)

• Now that this is created we will need to set up the following line in the main.py file

# Import the models.py file
from . import models

# Import the engine and SessionLocal form database.py
from app.database import engine, SessionLocal

# Set up the creation
models.Base.metadata.create_all(bind=engine)

# We will additionally have to import the dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

• Now this should be good to go

• Every time we will interact with the database we will get a session

• Once we are done interacting, the session will close

• We can start setting up paths to our App using this code

# We will have to import Depeds from FastAPI
from fastapi import Depends, FastAPI, Response, status, HTTPException

# We also need to import Session from SQLAlchemy
from sqlalchemy.orm import Session

# Testing SQLAlchemy
@app.get("sqlalchemy")
def test_post(db: Session = Depends(get_db)):
    return {"status": "success"}

• By setting all of this up, SQLAlchemy will connect to the Database

• Check if the tables exist

• If they don't it will create them

• If they do it will mode on

• But now if we connect to the database we should see the table created from scratch

Cleanup

• Just to keep everything clean, we will move the dependency code in our database.py file

• This keeps everything clean, all the database stuff with the database stuff

• Additionally we will have to import this in our main.py file

database.py

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

# Template:
SQLALCHEMY_DATABASE_URL = 'postgresql://<username>:<password>@<ip-address/hostname>/<databasename>'

# Engine for database
engine = create_engine(SQLALCHEMY_DATABASE_URL)

# Session to interact with said database
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base for defining Database models
Base = declarative_base()

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

main.py

from app.database import engine, get_db

from fastapi import Depends, FastAPI, Response, status, HTTPException
from typing import Optional
from pydantic import BaseModel
from random import randrange

import time
import psycopg

from sqlalchemy.orm import Session
from . import models

# Removed SessionLocal and imported get_db function
from app.database import engine, get_db

models.Base.metadata.create_all(bind=engine)


app = FastAPI()

class Post(BaseModel):
    title: str
    content: str
    published: bool = True
    
# Testing SQLAlchemy
@app.get("sqlalchemy")
def test_post(db: Session = Depends(get_db)):
    return {"status": "success"}