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
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"}
