SQLAlchemy is a powerful and flexible SQL toolkit and Object Relational Mapper (ORM) for Python. Learning it from scratch to advanced involves understanding both the Core (SQL Expression Language) and the ORM layer.

What is SQLAlchemy?

SQLAlchemy is a popular SQL toolkit and ORM for Python. It provides:

• Full power of SQL through its Core (SQL Expression Language)
• High-level ORM for mapping classes to tables
• Support for multiple databases like PostgreSQL, MySQL, SQLite, Oracle, etc.

Learn SQLAlchemy from Scratch to Advanced

1. Basic Setup & Installation

pip install sqlalchemy

(Optional for SQLite)

2. SQLAlchemy Core (SQL Expression Language)

Learn this first to understand how SQLAlchemy builds SQL queries under the hood.

Concepts:

• Engine & Connection
• Metadata
• Table objects
• Insert, Select, Update, Delete

from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData

engine = create_engine('sqlite:///example.db')
metadata = MetaData()

users = Table(
    'users', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String),
)

metadata.create_all(engine)

# Insert data
with engine.connect() as conn:
    conn.execute(users.insert().values(name='Alice'))

3. SQLAlchemy ORM – Introduction

ORM maps Python classes to database tables.

Concepts:

• Declarative Base
• Sessions
• CRUD operations
• Querying with ORM

Example

from sqlalchemy.orm import declarative_base, sessionmaker
from sqlalchemy import Column, Integer, String, create_engine

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)

engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

user = User(name='Bob')
session.add(user)
session.commit()

4. Intermediate ORM Topics

• Query filtering: .filter(), .filter_by()
• Relationships: one-to-many, many-to-one, many-to-many
• Lazy/eager loading
• Cascading deletes
• Transactions
• Composite keys

5. Advanced ORM Topics

• Custom column types
• Hybrid properties
• Events and listeners
• Using raw SQL with ORM
• Inheritance mapping
• Session scoping strategies
• Versioning & optimistic concurrency

6. Async SQLAlchemy (v1.4+)

If you're using Python asyncio:

from sqlalchemy.ext.asyncio import create_async_engine

Explore the async engine, async session, and integration with async web frameworks.

7. Integration with Web Frameworks

• Flask-SQLAlchemy
• FastAPI + SQLAlchemy
• Django (Note: Django uses its own ORM, not SQLAlchemy)

8. Tools & Best Practices

• Alembic for migrations
• Connection pooling
• Testing with in-memory SQLite
• Dependency injection for sessions

#db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
DATABASE_URL="sqlite:///sqlite.db"
# creating db engine
engine=create_engine(url=DATABASE_URL,echo=True)
'''
if echo=True it will log on the terminal what is happening in behind all 
sql query will be og in terminal
'''
SessionLocal=sessionmaker(bind=engine,expire_on_commit=False)

# models.py 
from db import engine
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column
from sqlalchemy import String,Integer
class Base(DeclarativeBase):
    '''base  declarative model for the other tables'''
    pass

class User(Base):
    '''User Table to store all teh information related to user'''
    __tablename__='USERS'
    id:Mapped[int]=mapped_column(primary_key=True,unique=True)
    name:Mapped[str]=mapped_column(String(50),nullable=False)
    email:Mapped[str]=mapped_column(String(50),nullable=False,unique=True)
    phone:Mapped[str]=mapped_column(String(50),nullable=False,unique=True)
    age:Mapped[int]=mapped_column(Integer,nullable=True)

    def __repr__(self):
        return f"User : -> {self.name}"
    
def create_table():
    '''creting all the tables related to the provided engine'''
    print("tables created successfully......")
    Base.metadata.create_all(engine)
    
#main.py
from fastapi import FastAPI
from db import SessionLocal
from models import create_table
from sqlalchemy import select
from models import User
app=FastAPI()
# create_table()
@app.get('/')
async def get_users():
    with SessionLocal() as session:
        stmt=select(User)
        users=session.scalars(stmt).all()
        return users
@app.post('/')
async def get_users(name:str,email:str,phone:str):
    with SessionLocal() as session:
        user=User(name=name,email=email,phone=phone)
        session.add(user)
        session.commit()
        return {"message":"user added successfully"}