A SQL Adapter allows you to connect and manipulate SQL Databases in a specific programming language. Object Relational Mapper abstracts away the need for SQL code in the project and is an additional layer on top of an 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 is some code examples that demonstrate the differnce between SQL developement in Python using an Adapter and SQL over modeling and mapping using ORM SQL Alchemy:
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 succesful")
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
# 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)):
# 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()
new_post = models.Post(**post.dict())
db.add(new_post)
db.commit()
db.refresh(new_post)
return new_post
PEP 249 Python Database API Specification v2.0 link
Psycopg SQL Adapterlink
SQL Alchemy ORM and SQL Toolkitlink
pydanticlink