FastAPI Integration

Integrate Cipherion encryption into your FastAPI applications store sensitive data securely and decrypt it on retrieval.

This guide walks you through integrating Cipherion into a FastAPI application. You will learn how to:

Configure the Cipherion client
Encrypt sensitive fields before writing to a database
Decrypt fields when returning records to the client
Structure your project with a clean service layer

Prerequisites

Install the required packages:

pip install fastapi uvicorn python-dotenv cipherion sqlalchemy

Project Structure

fastapi-cipherion-app/
│
├── app/
│   ├── config/
│   │   └── cipherion.py        # Cipherion client setup
│   ├── database/
│   │   └── db.py               # SQLAlchemy engine + session
│   ├── models/
│   │   └── models.py           # ORM model
│   ├── routes/
│   │   └── example_db.py       # CRUD routes
│   ├── schemas/
│   │   └── user.py             # Pydantic schemas
│   └── services/
│       └── encryption_service.py  # Encryption logic
│
├── main.py
├── demo.py
└── .env

Setup

Configure Environment Variables

Create a .env file in your project root:

CIPHERION_BASE_URL=https://api.cipherion.in
CIPHERION_PROJECT_ID=your_project_id
CIPHERION_API_KEY=your_api_key
CIPHERION_PASSPHRASE=your_passphrase

Initialize the Cipherion Client

app/config/cipherion.py

from cipherion import CipherionClient
import os
from dotenv import load_dotenv

load_dotenv()

cipherion_client = CipherionClient({
    "base_url": os.getenv("CIPHERION_BASE_URL"),
    "project_id": os.getenv("CIPHERION_PROJECT_ID"),
    "api_key": os.getenv("CIPHERION_API_KEY"),
    "passphrase": os.getenv("CIPHERION_PASSPHRASE"),
    "log_level": "info",
})

Create the FastAPI App

main.py

from fastapi import FastAPI
from app.routes import example_db

app = FastAPI(title="Cipherion FastAPI Integration")

app.include_router(example_db.router)

@app.get("/")
def health():
    return {"status": "ok"}

Quick Encryption Example

Before building the full CRUD API, let's verify your setup with a simple encryption test.

demo.py

from app.services.encryption_service import encryption_service

payload = {
    "username": "john",
    "email": "john@example.com",
    "phone": "9876543210",
    "role": "user",
}

encrypted = encryption_service.encrypt_user_data(payload)

print("Encrypted Data:", encrypted["encrypted"])
# {
#   "username": "john",         # excluded — not encrypted
#   "email":    "7f1cd70f...",  # encrypted
#   "phone":    "88c589cb...",  # encrypted
#   "role":     "user"          # excluded — not encrypted
# }

Run it:

python demo.py

username and role are plain text because they are listed in exclude_fields inside the service layer. Only email and phone are encrypted.

Database

Engine & Session

app/database/db.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "sqlite:///./users.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

ORM Model

app/models/models.py

from sqlalchemy import Column, String, Boolean
from app.database.db import Base


class User(Base):
    __tablename__ = "users"

    id        = Column(String, primary_key=True, index=True)
    username  = Column(String)
    email     = Column(String)   # stored encrypted
    phone     = Column(String)   # stored encrypted
    role      = Column(String)
    encrypted = Column(Boolean, default=True)

email and phone are stored as cipher text in the database. Never query these columns with a raw value — use the username, id, or other plaintext fields for lookups.

Encryption Service

Centralizing encryption logic keeps routes simple and ensures consistent exclusion rules across all operations.

app/services/encryption_service.py

import time
from typing import Any
from cipherion import CipherionError, DeepEncryptOptions, DeepDecryptOptions
from app.config.cipherion import cipherion_client


class EncryptionService:

    def __init__(self):
        # These fields are excluded from encryption so they remain
        # queryable and human-readable in the database.
        self.exclude_options = DeepEncryptOptions(
            exclude_fields=["_id", "username", "role"],
            exclude_patterns=["*_at", "*_id"],
        )

    def encrypt_user_data(self, data: Any):
        try:
            return cipherion_client.deep_encrypt(data, self.exclude_options)
        except CipherionError as error:
            if hasattr(error, "is_retryable") and error.is_retryable():
                time.sleep(2)
                return cipherion_client.deep_encrypt(data, self.exclude_options)
            raise

    def decrypt_user_data(self, data: Any):
        # Decryption MUST use the same exclusions that were used during encryption.
        return cipherion_client.deep_decrypt(
            data,
            DeepDecryptOptions(
                exclude_fields=["_id", "username", "role"],
                exclude_patterns=["*_at", "*_id"],
                fail_gracefully=True,
            ),
        )


encryption_service = EncryptionService()

Always pass the same exclude_fields and exclude_patterns to deep_decrypt that you used in deep_encrypt. Mismatched exclusions can cause Cipherion to attempt decryption on plaintext values, corrupting your data. See Decryption — Use Same Exclusions.

CRUD Routes

app/routes/example_db.py

Create User

Sensitive fields are encrypted before the record is written.

from fastapi import APIRouter, HTTPException, Depends
from sqlalchemy.orm import Session
from app.schemas.user import UserCreate, UserUpdate
from app.services.encryption_service import encryption_service
from app.database.db import get_db
from app.models.models import User

router = APIRouter(prefix="/users", tags=["Users"])


@router.post("/")
def create_user(payload: UserCreate, db: Session = Depends(get_db)):
    existing = db.query(User).filter(User.username == payload.username).first()
    if existing:
        raise HTTPException(status_code=400, detail="User already exists")

    encrypted = encryption_service.encrypt_user_data(payload.dict())
    encrypted["encrypted"].pop("encrypted", None)   # strip any stray flags

    user = User(
        id=payload.username,
        username=payload.username,
        email=encrypted["encrypted"].get("email"),  # cipher text
        phone=encrypted["encrypted"].get("phone"),  # cipher text
        role=payload.role,
    )

    db.add(user)
    db.commit()
    db.refresh(user)

    return {"success": True, "user": user.id}

Get User

The record is fetched and decrypted before being returned to the client.

@router.get("/{user_id}")
def get_user(user_id: str, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    encrypted_data = {
        "_id": user.id,
        "username": user.username,
        "email": user.email,
        "phone": user.phone,
        "role": user.role,
        "_encrypted": True,
    }

    decrypted = encryption_service.decrypt_user_data(encrypted_data)
    data = decrypted["data"]
    data.pop("_encrypted", None)

    return {"success": True, "data": data}

Update User

Only modified sensitive fields are re-encrypted plaintext fields are updated directly.

@router.put("/{user_id}")
def update_user(user_id: str, updates: UserUpdate, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    update_data = {k: v for k, v in updates.dict().items() if v is not None}
    if not update_data:
        raise HTTPException(status_code=400, detail="No fields to update")

    encrypted = encryption_service.encrypt_user_data(update_data)

    if "email" in encrypted["encrypted"]:
        user.email = encrypted["encrypted"]["email"]
    if "phone" in encrypted["encrypted"]:
        user.phone = encrypted["encrypted"]["phone"]
    if "role" in update_data:
        user.role = update_data["role"]

    db.commit()
    db.refresh(user)

    # Decrypt before returning the updated record
    encrypted_data = {
        "_id": user.id,
        "username": user.username,
        "email": user.email,
        "phone": user.phone,
        "role": user.role,
        "_encrypted": True,
    }

    decrypted = encryption_service.decrypt_user_data(encrypted_data)
    data = decrypted["data"]
    data.pop("_encrypted", None)

    return {"success": True, "message": "User updated", "data": data}

List Users

Each record is decrypted individually before building the response list.

@router.get("/")
def list_users(db: Session = Depends(get_db)):
    users = db.query(User).all()
    results = []

    for user in users:
        encrypted_data = {
            "_id": user.id,
            "username": user.username,
            "email": user.email,
            "phone": user.phone,
            "role": user.role,
            "_encrypted": True,
        }

        decrypted = encryption_service.decrypt_user_data(encrypted_data)
        data = decrypted["data"]
        data.pop("_encrypted", None)
        results.append(data)

    return {"success": True, "count": len(results), "users": results}

For large datasets, consider parallel decryption to reduce response time instead of decrypting records sequentially.

End-to-End Flow

POST /users
   │
   ▼
FastAPI Route
   │  payload = { username, email, phone, role }
   ▼
EncryptionService.encrypt_user_data()
   │  deep_encrypt — email & phone become cipher text
   │  username & role stay plaintext (excluded)
   ▼
SQLite (users table)
   │  email = "7f1cd70f...", phone = "88c589cb..."
   ▼
GET /users/{id}
   │
   ▼
EncryptionService.decrypt_user_data()
   │  deep_decrypt — cipher text → original values
   ▼
JSON Response
   │  { email: "john@example.com", phone: "9876543210", ... }

Running the Server

uvicorn main:app --reload

Open the interactive docs at http://localhost:8000/docs.

For a deeper understanding of encryption, decryption, and field exclusions, refer to the Core Concepts section.

FastAPI Integration

On this page