SQL Driver vs ORM

5th January 2023 · sql, python, 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

References