Database & ORM

Caspian provides a type-safe, auto-generated Async Python client for your database. Follow this guide to configure your environment, define your schema, and perform non-blocking operations.

Environment Setup

Define your database connection string in the .env file. This is the entry point for the Prisma engine.

.env

# Connect to SQLite (Recommended for dev)
DATABASE_URL="file:./prisma/dev.db"

# Or PostgreSQL / MySQL (Recommended for Async/Production)
# DATABASE_URL="postgresql://user:password@localhost:5432/mydb"

Global Configuration

Use prisma.config.ts to configure seeding behavior and global ORM settings.

prisma.config.ts

export default 
  seed: 
    import: "./prisma/seed.ts",
    autoRun: true
  ,
  client: 
    logLevel: "info"

Define Your Schema

Model your data in prisma/schema.prisma. This single file acts as the source of truth.

prisma/schema.prisma

datasource db 
  provider = "sqlite"


model User 
  id        String   @id @default(cuid())
  email     String   @unique
  name      String?
  posts     Post[]


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

Command Reference

Use the CLI to sync, migrate, and seed your database.

Command	Description
npx ppy generate	Critical Compiles the Prisma schema into a type-safe Python client. Run this after every schema change.
npx prisma migrate dev	Creates a new SQL migration file based on your schema changes and applies it to the development database.
npx prisma migrate deploy	Applies pending migrations to production databases. Use this in your CI/CD pipelines.
npx prisma db push	Quickly syncs the database schema without creating a migration file. Great for prototyping.
npx prisma db seed	Populates the database with initial data defined in your seeding script (configured in `prisma.config.ts`).
npx prisma studio	Opens a visual GUI in your browser to inspect, edit, and manage database records manually.

Async Client Usage

Import the global prisma instance. It is pre-configured and ready to use. The client is fully asynchronous, so always use await for database operations.

Ready-to-Use Import

from src.lib.prisma.db import prisma

# No manual connection required.
# Just import and await.

Non-Blocking I/O

Queries run on the event loop, allowing your application to handle concurrent requests efficiently. Ideal for high-performance RPC agents and endpoints.

Create

user = await prisma.user.create(
    data=
        'email': "alice@example.com",
        'name': "Alice",
    
)

Read

Filtering & Sorting

posts = await prisma.post.find_many(
    where=
        'published': True,
        'title': 'contains': "Caspian"
    ,
    order_by=
        'createdAt': 'desc'
    
)

Selection & Relations

# Fetch user and join their role data
user = await prisma.user.find_unique(
    where=
        'email': "alice@example.com"
    ,
    include=
        'userRole': True 
    
)

Update

Atomic updates are supported for numeric fields.

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

Delete

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

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

Create

user = await prisma.user.create(
    data=
        'email': "alice@example.com",
        'name': "Alice",
    
)

Read

Filtering & Sorting

posts = await prisma.post.find_many(
    where=
        'published': True,
        'title': 'contains': "Caspian"
    ,
    order_by=
        'createdAt': 'desc'
    
)

Selection & Relations

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

Update

Atomic updates are supported for numeric fields.

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

Delete

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

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

Advanced Features

Aggregations

Perform calculations directly in the database using aggregate or group_by.

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

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

Async Transactions

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

async with prisma.tx() as tx:
    user = await tx.user.create(data=...)
    await tx.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 = await prisma.query_raw(
    "SELECT * FROM User WHERE email LIKE ?", 
    "%@gmail.com"
)