Flowfull Docs

Appearance

Flowfull-Python ​

Official Library + Starter Kit

Python has two official pieces:

Use flowfull-python when you need the Python client package in your own backend. Use flowfull-python-starter when you want a complete FastAPI backend template already wired with Flowfull concepts.

Status ​

Available - official Python library plus production-shaped starter for Python/FastAPI-style teams.

Install the library:

bash
pip install flowfull-python

Or clone the starter:

bash
git clone https://github.com/pubflow/flowfull-python-starter.git my-backend
cd my-backend
python -m venv venv
source venv/bin/activate
pip install -r requirements.txt

The Python starter keeps the same Flowfull model: Bridge Validation, auth middleware, portable database patterns, and custom backend routes.

Included / Expected Features ​

  • ✅ Python 3.11+ - Modern Python with type hints
  • ✅ FastAPI - Fast, modern web framework
  • ✅ SQLAlchemy or Tortoise ORM - Database abstraction
  • ✅ Pydantic - Data validation
  • ✅ All 7 Core Concepts - Bridge Validation, HybridCache, Trust Tokens, etc.
  • ✅ Type Safety - Full type hints with mypy
  • ✅ Async/Await - Asynchronous request handling

Starter Structure ​

flowfull-python-starter/
├── app/
│   ├── lib/
│   │   ├── bridge/           # Bridge Validation
│   │   ├── cache/            # HybridCache
│   │   ├── auth/             # Auth Middleware
│   │   ├── database/         # Database
│   │   ├── tokens/           # Trust Tokens (PASETO)
│   │   └── config/           # Configuration
│   ├── routes/               # API Routes
│   │   ├── auth.py
│   │   └── users.py
│   └── main.py               # Entry Point
├── migrations/               # Database Migrations (Alembic)
├── tests/
├── .env.example
├── requirements.txt
├── pyproject.toml
└── README.md

Technology Stack ​

TechnologyPurposeVersion
PythonLanguage3.11+
FastAPIWeb Framework0.100+
SQLAlchemyORM2.0+
PydanticValidation2.0+
redis-pyRedis ClientLatest
pysetoTrust TokensLatest
python-dotenvEnvironmentLatest

Core Concepts (Preview) ​

1. Bridge Validation ​

python
from typing import Optional
from dataclasses import dataclass
import httpx

@dataclass
class ValidationOptions:
    ip: Optional[str] = None
    user_agent: Optional[str] = None
    device_id: Optional[str] = None

class BridgeValidator:
    def __init__(self, flowless_url: str, bridge_secret: str):
        self.flowless_url = flowless_url
        self.bridge_secret = bridge_secret
    
    async def validate_session(
        self, 
        session_id: str, 
        opts: ValidationOptions
    ) -> dict:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.flowless_url}/api/bridge/validate",
                json={
                    "session_id": session_id,
                    "bridge_secret": self.bridge_secret,
                    "ip": opts.ip,
                    "user_agent": opts.user_agent,
                    "device_id": opts.device_id
                }
            )
            response.raise_for_status()
            return response.json()

2. HybridCache ​

python
from typing import Optional, Callable, Any
from functools import lru_cache
import redis.asyncio as redis

class HybridCache:
    def __init__(self, redis_url: str, max_size: int = 1000):
        self.redis = redis.from_url(redis_url)
        self.max_size = max_size
    
    async def get(
        self, 
        namespace: str, 
        key: str, 
        fetcher: Callable[[], Any]
    ) -> Any:
        cache_key = f"{namespace}:{key}"
        
        # Layer 1: LRU (in-memory)
        # Layer 2: Redis
        cached = await self.redis.get(cache_key)
        if cached:
            return cached
        
        # Layer 3: Database
        value = await fetcher()
        
        # Store in Redis
        await self.redis.setex(cache_key, 300, value)
        
        return value
    
    async def invalidate(self, namespace: str, key: str):
        cache_key = f"{namespace}:{key}"
        await self.redis.delete(cache_key)

3. Auth Middleware ​

python
from fastapi import Request, HTTPException, Depends
from typing import Optional

async def require_auth(request: Request) -> str:
    """Middleware to require authentication"""
    session_id = request.headers.get("X-Session-Id")
    
    if not session_id:
        raise HTTPException(status_code=401, detail="Unauthorized")
    
    validator = request.app.state.bridge_validator
    
    try:
        session = await validator.validate_session(
            session_id,
            ValidationOptions(
                ip=request.client.host,
                user_agent=request.headers.get("User-Agent")
            )
        )
        return session["user_id"]
    except Exception:
        raise HTTPException(status_code=401, detail="Invalid session")

async def optional_auth(request: Request) -> Optional[str]:
    """Middleware for optional authentication"""
    try:
        return await require_auth(request)
    except HTTPException:
        return None

4. FastAPI Route Example ​

python
from fastapi import FastAPI, Depends
from app.lib.auth.middleware import require_auth
from app.lib.cache.hybrid_cache import HybridCache

app = FastAPI()

@app.get("/api/profile")
async def get_profile(
    user_id: str = Depends(require_auth),
    cache: HybridCache = Depends()
):
    user = await cache.get("user", user_id, lambda: fetch_user_from_db(user_id))
    return {"user": user}

Repository ​

Resources ​

Support ​