Flowfull-Python

Community Starter Kit 🚧 Coming Soon

A Python implementation of Flowfull is currently in development by the community.

Status

🚧 In Development - Community-driven starter kit

The Python implementation will include all 7 core concepts adapted for Python's ecosystem and best practices.

Planned 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

Planned Structure

flowfull-python/
├── 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 (Planned)

Technology	Purpose	Version
Python	Language	3.11+
FastAPI	Web Framework	0.100+
SQLAlchemy	ORM	2.0+
Pydantic	Validation	2.0+
redis-py	Redis Client	Latest
pyseto	Trust Tokens	Latest
python-dotenv	Environment	Latest

Core Concepts (Preview)

1. Bridge Validation

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

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

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

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}

Want to Contribute?

We're looking for Python developers to help build the official Python starter kit!

How to Contribute

1. Join the discussion - discord.gg/pubflow
2. Review the 7 Core Concepts - Core Concepts
3. Implement in Python - Follow Python best practices (PEP 8, type hints)
4. Submit PR - github.com/pubflow/flowfull-python

Requirements

• Implement all 7 core concepts
• Follow Python conventions (PEP 8, PEP 484)
• Include comprehensive tests (pytest)
• Provide clear documentation
• Support Python 3.11+
• Full type hints with mypy validation

Resources

• Core Concepts: 7 Core Concepts
• Official Node.js Implementation: flowfull-node
• Community Contributions: Getting Started - Community Contributions

Support

• Community: discord.gg/pubflow
• Email: support@pubflow.com
• Professional Support: notside.com - contact@notside.com

---

Interested in building this? Join us! 🚀