Database & ORM

Caspian includes a high-performance, auto-generated Python client based on your Prisma schema. It translates Prisma syntax directly into optimized SQL, eliminating the need for the heavy Prisma Engine binary.

Defining Your Data

Your database structure is defined in prisma/schema.prisma. Caspian uses this single source of truth to generate your SQL migrations and your type-safe Python client.

prisma/schema.prisma

model User {
  id        String   @id @default(cuid())
  email     String   @unique
  name      String?
  posts     Post[]
  createdAt DateTime @default(now())
}

model Post {
  id        String   @id @default(cuid())
  title     String
  published Boolean  @default(false)
  author    User     @relation(fields: [authorId], references: [id])
  authorId  String
}

The Prisma Client

Import the global prisma instance to interact with your database. The client is fully typed, providing autocomplete for all your models and fields.

Basic Usage

from lib.prisma.db import prisma

# The client is synchronous
users = prisma.user.find_many()

Connection Pooling

The client automatically manages a thread-safe connection pool for SQLite, ensuring performance under load in multithreaded environments like FastAPI.

CRUD Operations

Create

Create new records using create or create_many.

# Create a single record
user = prisma.user.create(
    data={
        'email': "alice@example.com",
        'name': "Alice",
    }
)

# Create multiple records (Atomic batch insert)
prisma.todo.create_many(
    data=[
        {'title': "Buy milk"},
        {'title': "Walk dog"},
    ],
    skip_duplicates=True
)

Read

Retrieve data with find_many, find_unique, or find_first. The API supports rich filtering and pagination options.

Filtering & Sorting

posts = prisma.post.find_many(
    where={
        'published': True,
        'title': {'contains': "Caspian"}  # LIKE %Caspian%
    },
    order_by={
        'createdAt': 'desc'
    }
)

Selection & Relations

# Fetch user and join their role data
user = prisma.user.find_unique(
    where={
        'email': "alice@example.com"
    },
    include={
        'userRole': True 
    },
    # OR specific fields
    select={
        'id': True,
        'name': True
    }
)

Update

Atomic updates are supported for numeric fields.

prisma.user.update(
    where={'id': "123"},
    data={
        'name': "New Name",
        # Atomic increment
        'loginCount': {'increment': 1}
    }
)

Delete

# Delete specific record
prisma.user.delete(where={'id': "123"})

# Batch delete
prisma.log.delete_many(
    where={
        'createdAt': {'lt': yesterday}
    }
)

Advanced Features

Aggregations

Perform calculations directly in the database using aggregate or group_by.

stats = prisma.sales.aggregate(
    _sum={'amount': True},
    _avg={'rating': True},
    where={'status': "completed"}
)

print(stats['_sum']['amount'])

Transactions

Ensure data integrity with atomic transactions. If any operation fails, all are rolled back.

with prisma.transaction():
    user = prisma.user.create(data={...})
    prisma.log.create(data={
        'action': "CREATED_USER",
        'userId': user.id
    })

Raw SQL Fallback

For complex queries that cannot be expressed via the ORM, use query_raw or execute_raw.

# Safe parameterized raw query
users = prisma.query_raw(
    "SELECT * FROM User WHERE email LIKE ?", 
    "%@gmail.com"
)