SQL Driver vs ORM
A SQL Adapter allows you to connect and manipulate SQL Databases in a specific programming language. An Object Relational Mapper abstracts away the need for SQL code in the project and is an additional layer on top of a SQL Adapter, but I would argue the need to have a good understanding of SQL is still required when using ORM and doesn't allow the developer to just rely on object mapping alone.
Below are code examples that demonstrate the difference between SQL development in Python using an Adapter and object modeling and mapping using ORM SQLAlchemy:
SQL Adapter (psycopg)
import psycopg
from psycopg import sql
from psycopg.rows import dict_row
while True:
try:
conn = psycopg.connect(host='**', dbname='**', user='**', password='**', row_factory=dict_row)
cursor = conn.cursor()
print("Database connection was successful")
break
except Exception as error:
print("Failed to connect to database")
print("error: ", error)
time.sleep(5)
@app.post("/posts", status_code=status.HTTP_201_CREATED, response_model=schemas.Post)
def create_post(post: schemas.PostCreate):
cursor.execute('INSERT INTO posts (title, content, published) VALUES (%s, %s, %s) RETURNING *', (post.title, post.content, post.published))
new_post = cursor.fetchone()
conn.commit()
return new_post
ORM (SQLAlchemy)
# models.py (SQLAlchemy)
from sqlalchemy import Column, Integer, String, Boolean
from .database import Base
from sqlalchemy.sql.expression import text
from sqlalchemy.sql.sqltypes import TIMESTAMP
class Post(Base):
__tablename__ = "posts"
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)
created_at = Column(TIMESTAMP(timezone=True), nullable=False, server_default=text('now()'))
# schemas.py (pydantic)
from pydantic import BaseModel, EmailStr
from datetime import datetime
class PostBase(BaseModel):
title: str
content: str
published: bool = True
class PostCreate(PostBase):
pass
class Post(PostBase):
id: int
created_at: datetime
class Config:
orm_mode = True
# main.py
from fastapi import FastAPI, Response, status, HTTPException, Depends
from fastapi.params import Body
from pydantic import BaseModel
from typing import Optional, List
from random import randrange
import psycopg
from psycopg import sql
from psycopg.rows import dict_row
from sqlalchemy.orm import Session
import time
from . import models, schemas
from .database import engine, SessionLocal, get_db
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
@app.post("/posts", status_code=status.HTTP_201_CREATED, response_model=schemas.Post)
def create_post(post: schemas.PostCreate, db: Session = Depends(get_db)):
new_post = models.Post(**post.dict())
db.add(new_post)
db.commit()
db.refresh(new_post)
return new_post