ORM CRUD with SQLAlchemy, Prisma & Hibernate

Overview

An ORM (Object-Relational Mapper) turns table rows into language objects, so CRUD operations become method calls instead of handwritten SQL. This recipe builds the same User/Post model in three ORMs — SQLAlchemy (Python), Prisma (JavaScript), and Hibernate/JPA (Java) — and then shows variants for Django ORM, Spring Data, and TypeORM.

The pattern underneath is always the same: declare a model, let the framework translate your object operations into parameterized SQL, and get rows back as objects. The differences live in how each tool expresses relationships, transactions, and eager loading.

When to Use

Reach for an ORM when:

• Your app has several entity types with relationships (users, posts, orders) and you’re tired of hand-writing the same INSERT/SELECT/UPDATE/DELETE statements.
• You want type-safe queries and IDE autocomplete instead of SQL strings that fail at runtime.
• Schema changes should live next to the models — Alembic for SQLAlchemy, Prisma Migrate, or Flyway/Liquibase for Hibernate.
• Most of your queries are CRUD plus simple joins, which is the majority of application code.

Skip the ORM and use raw SQL for heavy aggregations, ETL jobs, or hot paths where you need to control the exact generated query — ORMs generate generic SQL that’s fine for CRUD but often wasteful for analytics.

Solution

Python — SQLAlchemy

from sqlalchemy import create_engine, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String, unique=True)
    role: Mapped[str] = mapped_column(String, default="user")

engine = create_engine("postgresql+psycopg://user:pass@localhost/mydb")

with Session(engine) as session:
    # Create
    user = User(email="alice@example.com", role="admin")
    session.add(user)
    session.commit()

    # Read
    user = session.scalar(
        select(User).where(User.email == "alice@example.com")
    )

    # Update — the session tracks the dirty attribute
    user.role = "superadmin"
    session.commit()

    # Delete
    session.delete(user)
    session.commit()

SQLAlchemy 2.x style: select() + session.scalar()/session.execute() replace the legacy session.query() API. The with Session(...) block guarantees the connection goes back to the pool even when something raises.

JavaScript — Prisma

const { PrismaClient } = require('@prisma/client');
const prisma = new PrismaClient();

async function crud() {
    // Create
    const user = await prisma.user.create({
        data: { email: 'alice@example.com', role: 'admin' }
    });

    // Read
    const found = await prisma.user.findUnique({
        where: { email: 'alice@example.com' }
    });

    // Update
    const updated = await prisma.user.update({
        where: { id: user.id },
        data: { role: 'superadmin' }
    });

    // Delete
    await prisma.user.delete({ where: { id: user.id } });
}

Java — Hibernate / JPA

import jakarta.persistence.*;
import java.util.List;

@Entity
@Table(name = "users")
public class User {
    @Id @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Integer id;
    @Column(nullable = false, unique = true)
    private String email;
    private String role = "user";

    // getters and setters omitted
}

public class UserRepository {
    private final EntityManager em;

    public UserRepository(EntityManager em) { this.em = em; }

    public void create(User user) {
        em.getTransaction().begin();
        em.persist(user);
        em.getTransaction().commit();
    }

    public User findByEmail(String email) {
        return em.createQuery("SELECT u FROM User u WHERE u.email = :email", User.class)
                 .setParameter("email", email)
                 .getSingleResult();
    }

    public void updateRole(Integer id, String role) {
        em.getTransaction().begin();
        User user = em.find(User.class, id);
        user.setRole(role);
        em.getTransaction().commit();
    }

    public void delete(Integer id) {
        em.getTransaction().begin();
        em.remove(em.find(User.class, id));
        em.getTransaction().commit();
    }
}

Explanation

The three tools share the same pipeline — your objects go in, SQL comes out, rows come back as objects:

SQLAlchemy maps classes to tables through DeclarativeBase and Mapped columns. A Session is both a unit of work and a cache: it tracks which objects changed and wraps those changes in a transaction when you call commit(). Prisma takes a different route — you declare models once in schema.prisma and it generates a fully typed client, so a mistyped field fails in your editor before it ever reaches the database. Hibernate uses JPA annotations (@Entity, @Id, @Column); the EntityManager is the persistence context that detects dirty entities and flushes them on commit.

None of this removes SQL — it just moves it. When a query gets slow, the ORM output is the first place to look, and checking the generated query with EXPLAIN beats guessing.

Variants

Technology	Approach	Notes
Python	Django ORM	Batteries-included, tightly coupled to Django
JavaScript	TypeORM	Decorator-based ORM with strong TypeScript support
Java	Spring Data JPA	Repository abstraction on top of Hibernate
MongoDB	Mongoose	Document store, not relational — see the Mongoose CRUD recipe

Advanced Examples

SQLAlchemy with Relationships and Eager Loading

from sqlalchemy import create_engine, String, ForeignKey, select, update
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship, selectinload

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String, unique=True)
    role: Mapped[str] = mapped_column(String, default="user")
    posts: Mapped[list["Post"]] = relationship(back_populates="author", cascade="all, delete-orphan")

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    author: Mapped[User] = relationship(back_populates="posts")

engine = create_engine("postgresql+psycopg://user:pass@localhost/mydb")

with Session(engine) as session:
    # Eager loading with selectinload — avoids N+1
    stmt = (
        select(User)
        .options(selectinload(User.posts))
        .where(User.role == "admin")
    )
    admins_with_posts = session.execute(stmt).scalars().all()

    # Bulk insert
    session.add_all([
        User(email=f"user{i}@example.com", role="user")
        for i in range(100)
    ])
    session.commit()

    # Bulk update with Core
    session.execute(
        update(User)
        .where(User.role == "user")
        .values(role="member")
    )
    session.commit()

Prisma Schema with Relations

// schema.prisma
model User {
  id    Int     @id @default(autoincrement())
  email String  @unique
  role  String  @default("user")
  posts Post[]

  @@index([role])
}

model Post {
  id       Int    @id @default(autoincrement())
  title    String
  authorId Int
  author   User   @relation(fields: [authorId], references: [id])

  @@index([authorId])
}

const { PrismaClient } = require('@prisma/client');
const prisma = new PrismaClient();

// Create with nested records
const user = await prisma.user.create({
  data: {
    email: 'alice@example.com',
    role: 'admin',
    posts: {
      create: [
        { title: 'First Post' },
        { title: 'Second Post' },
      ],
    },
  },
  include: { posts: true }, // Eager load
});

// Transaction with multiple operations
const [newUser, updatedPost] = await prisma.$transaction([
  prisma.user.create({ data: { email: 'bob@example.com' } }),
  prisma.post.update({ where: { id: 1 }, data: { title: 'Updated' } }),
]);

// Upsert (create or update)
const result = await prisma.user.upsert({
  where: { email: 'carol@example.com' },
  update: { role: 'admin' },
  create: { email: 'carol@example.com', role: 'admin' },
});

// Raw SQL for complex queries
const topUsers = await prisma.$queryRaw`
  SELECT u.email, COUNT(p.id) AS post_count
  FROM users u
  LEFT JOIN posts p ON p.author_id = u.id
  GROUP BY u.email
  ORDER BY post_count DESC
  LIMIT 10
`;

Django ORM

# models.py
from django.db import models

class User(models.Model):
    email = models.EmailField(unique=True)
    role = models.CharField(max_length=20, default='user')

    class Meta:
        indexes = [models.Index(fields=['role'])]

class Post(models.Model):
    title = models.CharField(max_length=200)
    author = models.ForeignKey(User, on_delete=models.CASCADE, related_name='posts')

# CRUD operations
user = User.objects.create(email='alice@example.com', role='admin')
users = User.objects.filter(role='admin').prefetch_related('posts')  # Eager load
user.role = 'superadmin'
user.save()
user.delete()

# Bulk create
User.objects.bulk_create([
    User(email=f'user{i}@example.com') for i in range(100)
])

# Bulk update
User.objects.filter(role='user').update(role='member')

# Aggregation
from django.db.models import Count
User.objects.annotate(post_count=Count('posts')).filter(post_count__gt=5)

Spring Data JPA Repository

public interface UserRepository extends JpaRepository<User, Integer> {

    // Derived query methods
    Optional<User> findByEmail(String email);
    List<User> findByRole(String role);

    // Custom queries with @Query
    @Query("SELECT u FROM User u WHERE u.role = :role ORDER BY u.email")
    List<User> findByRoleOrdered(@Param("role") String role);

    // Native query
    @Query(value = "SELECT * FROM users WHERE email LIKE :pattern",
           nativeQuery = true)
    List<User> findByEmailPattern(@Param("pattern") String pattern);

    // Modifying queries
    @Modifying
    @Query("UPDATE User u SET u.role = :role WHERE u.id = :id")
    int updateRole(@Param("id") Integer id, @Param("role") String role);

    // Pagination and sorting
    Page<User> findByRole(String role, Pageable pageable);
}

// Usage with pagination
Pageable pageable = PageRequest.of(0, 20, Sort.by("email").ascending());
Page<User> adminPage = userRepository.findByRole("admin", pageable);

TypeORM (TypeScript)

import { Entity, PrimaryGeneratedColumn, Column, OneToMany, ManyToOne } from 'typeorm';

@Entity()
class User {
  @PrimaryGeneratedColumn()
  id: number;

  @Column({ unique: true })
  email: string;

  @Column({ default: 'user' })
  role: string;

  @OneToMany(() => Post, post => post.author)
  posts: Post[];
}

@Entity()
class Post {
  @PrimaryGeneratedColumn()
  id: number;

  @Column()
  title: string;

  @ManyToOne(() => User, user => user.posts)
  author: User;
}

// Repository pattern
const userRepo = dataSource.getRepository(User);

// Create
const user = userRepo.create({ email: 'alice@example.com', role: 'admin' });
await userRepo.save(user);

// Read with relations
const users = await userRepo.find({
  where: { role: 'admin' },
  relations: ['posts'],
  order: { email: 'ASC' },
  take: 20,
  skip: 0,
});

// Bulk operations
await userRepo
  .createQueryBuilder()
  .update()
  .set({ role: 'member' })
  .where('role = :role', { role: 'user' })
  .execute();

What Works

1. Define explicit schemas and constraints in the ORM, not just the database — the model is documentation that doesn’t rot.
2. Wrap multi-step operations in transactions so a failed step doesn’t leave half-written state.
3. Add database-level indexes on the columns you filter or join on; the ORM won’t invent them for you.
4. Reach for eager loading (selectinload, include, JOIN FETCH) when you know you will touch a relationship — N+1 is the default failure mode, not the exception.
5. Keep entities thin: persistence details in the model, business rules in a service layer.

Common Mistakes

1. Forcing analytical or reporting queries through the ORM — generated SQL for complex aggregations is usually mediocre; drop to raw SQL for those.
2. Loading relationships inside loops and discovering the N+1 query problem in production.
3. Putting business logic inside entity classes, which makes the entities untestable without a database.
4. Reading lazily-loaded attributes outside a Hibernate session and hitting LazyInitializationException.
5. Never closing sessions or Prisma clients — each leak holds a pooled connection hostage.

Performance Tips

1. Use lean() in Mongoose or DTO projections in Hibernate. Skip ORM hydration for read-only queries:

// Mongoose: returns plain objects, not Mongoose documents
const users = await User.find().lean().exec();

// Hibernate DTO projection
@Query("SELECT new com.app.dto.UserDTO(u.id, u.email) FROM User u WHERE u.role = :role")
List<UserDTO> findDTOsByRole(@Param("role") String role);

2. Use cursor-based pagination instead of offset. Offset pagination gets slower as you go deeper:

# Bad: offset pagination (slow for large offsets)
session.scalars(select(User).offset(100000).limit(20)).all()

# Good: cursor pagination (constant time)
last_id = 100000
session.scalars(
    select(User).where(User.id > last_id).order_by(User.id).limit(20)
).all()

3. Enable second-level cache in Hibernate. Cache frequently read entities:

@Entity
@Cache(usage = CacheConcurrencyStrategy.READ_WRITE)
public class User {
    // ...
}

4. Use EXPLAIN ANALYZE on ORM-generated queries. The SQL your ORM generates may not use indexes efficiently. Always verify with EXPLAIN:

EXPLAIN ANALYZE SELECT users.id, users.email, users.role
FROM users
WHERE users.role = 'admin'
ORDER BY users.email;

5. Disable autoflush during bulk operations. SQLAlchemy flushes the session before every query by default, which is wasted work during batch inserts:

with session.no_autoflush:
    session.add_all(bulk_users)

See Also

• Runnable examples in the companion repository — the same User/Post CRUD in five ORMs, ready to execute.
• SQLAlchemy 2.x ORM quickstart — sessions, select(), and relationship loading strategies.
• Prisma CRUD reference — every create/find/update/delete variant, transactions, and raw queries.
• Hibernate ORM User Guide — persistence context, fetching, and caching.
• Django querysets — select_related vs prefetch_related and aggregation.