soleprint/artery/veins/PATTERNS.md

Vein Patterns

This document describes the patterns that emerged from building Jira, Slack, and Google veins side-by-side.

Philosophy

Core = Isolated API client - Can run without FastAPI, framework-agnostic Vein = Corset/wrapper - Makes the core speak to pawprint ecosystem

The vein wrapper is not a literal folder - it's following structural conventions and patterns.

Directory Structure (Standard)

vein/{service}/
├── core/                   # ISOLATED - can run standalone
│   ├── __init__.py
│   ├── config.py           # Pydantic settings from .env
│   ├── auth.py             # Auth logic (optional, for complex auth)
│   ├── client.py           # Main API client
│   └── {domain}.py         # Additional clients (sheets, drive, etc.)
│
├── api/                    # WRAPPER - FastAPI integration
│   ├── __init__.py
│   └── routes.py           # APIRouter with endpoints
│
├── models/                 # Data models
│   ├── __init__.py
│   ├── {domain}.py         # Pydantic models with from_{service}()
│   └── formatter.py        # Text formatters for LLM output
│
├── storage/                # Persistent data (optional, for OAuth tokens)
│   └── .gitignore
│
├── main.py                 # FastAPI app setup
├── run.py                  # Standalone runner
├── requirements.txt        # Dependencies
├── .env.example            # Configuration template
└── README.md              # Service-specific docs

Base Classes

BaseVein (vein/base.py)

Minimal interface for all veins:

• name: str - Service name
• get_client(creds) -> Client - Create API client
• health_check(creds) -> dict - Test connection

Used for simple token-based auth (Jira, Slack, WhatsApp).

BaseOAuthVein (vein/oauth.py)

Extends BaseVein for OAuth2 services:

• get_auth_url(state) -> str - Generate OAuth URL
• exchange_code(code) -> dict - Code for tokens
• refresh_token(refresh_token) -> dict - Refresh expired tokens
• storage: TokenStorage - Token persistence

Used for OAuth2 services (Google, GitHub, GitLab).

TokenStorage (vein/oauth.py)

File-based token storage (can be overridden for Redis/DB):

• save_tokens(user_id, tokens) - Persist tokens
• load_tokens(user_id) -> dict - Retrieve tokens
• is_expired(tokens) -> bool - Check expiration
• delete_tokens(user_id) - Logout

Core Module Patterns

config.py

from pathlib import Path
from pydantic_settings import BaseSettings

ENV_FILE = Path(__file__).parent.parent / ".env"

class {Service}Config(BaseSettings):
    # Service-specific settings
    api_port: int = 800X  # Unique port per vein

    model_config = {
        "env_file": ENV_FILE,
        "env_file_encoding": "utf-8",
    }

settings = {Service}Config()

Pattern: Pydantic BaseSettings with .env file at vein root.

client.py (Simple Auth - Jira, Slack)

from {sdk} import Client

def get_client(credentials) -> Client:
    """Create authenticated client."""
    return Client(credentials)

Pattern: Simple factory function returning SDK client.

oauth.py (OAuth2 - Google)

class {Service}OAuth:
    """OAuth2 client."""

    def get_authorization_url(self, state=None) -> str:
        """Generate auth URL for user redirect."""

    def exchange_code_for_tokens(self, code: str) -> dict:
        """Exchange code for tokens."""

    def refresh_access_token(self, refresh_token: str) -> dict:
        """Refresh expired token."""

    def get_credentials(self, access_token, refresh_token=None):
        """Create SDK credentials from tokens."""

Pattern: OAuth client handles full flow, separate from API client.

API Module Patterns

routes.py

from fastapi import APIRouter, HTTPException, Query
from fastapi.responses import PlainTextResponse

router = APIRouter()

@router.get("/health")
async def health(creds = Depends(get_credentials)):
    """Test connection."""
    return {"status": "ok", "user": "..."}

@router.get("/resource")
async def get_resource(text: bool = False):
    """Get resource data."""
    # ... fetch data
    return _maybe_text(data, text, formatter)

Standard endpoints:

• /health - Connection test (required)
• Service-specific resources
• ?text=true query param for text output

Helper Functions

def _maybe_text(data, text: bool, formatter):
    """Return text or JSON based on query param."""
    if not text:
        return data
    return PlainTextResponse(formatter(data))

Pattern: Consistent text/JSON toggle across all veins.

Model Patterns

Domain Models

from pydantic import BaseModel

class Resource(BaseModel):
    id: str
    name: str
    # ... fields

    @classmethod
    def from_{service}(cls, raw: dict) -> "Resource":
        """Parse from service API response."""
        return cls(
            id=raw["id"],
            name=raw["name"],
            # ... mapping
        )

Pattern: Pydantic models with from_{service}() factory methods.

Formatters

def format_{resource}(resource: Resource) -> str:
    """Format resource as text (LLM-friendly)."""
    return f"{resource.name} (ID: {resource.id})"

Pattern: Simple functions returning plain text, no fancy tables.

Authentication Patterns

Simple Token Auth (Jira, Slack, WhatsApp)

Headers or .env fallback:

async def get_{service}_credentials(
    x_{service}_token: str | None = Header(None),
) -> Credentials:
    # Use header if provided
    if x_{service}_token and x_{service}_token.strip():
        return Credentials(token=x_{service}_token.strip())

    # Fall back to .env
    if settings.{service}_token:
        return Credentials(token=settings.{service}_token)

    raise HTTPException(401, "Missing credentials")

Pattern: Per-request headers for web UI, .env for standalone/API use.

OAuth2 (Google, GitHub, GitLab)

Three-step flow:

1. Start: GET /oauth/start → Redirect to service
2. Callback: GET /oauth/callback?code=... → Exchange code, save tokens
3. Use: Load tokens from storage, auto-refresh if expired

Pattern: Stateful (requires token storage), user must complete browser flow.

Error Handling

try:
    client = get_client(creds)
    data = client.fetch_something()
    return data
except {Service}ClientError as e:
    raise HTTPException(500, str(e))
except Exception as e:
    raise HTTPException(500, f"Unexpected error: {e}")

Pattern: Catch service-specific errors first, then generic fallback.

Configuration Files

.env.example

Include all required settings with placeholder values:

# Service credentials
SERVICE_API_KEY=your_key_here
SERVICE_URL=https://api.service.com

# Vein config
API_PORT=8001

requirements.txt

Minimal dependencies:

fastapi>=0.104.0
uvicorn>=0.24.0
pydantic>=2.0.0
pydantic-settings>=2.0.0
{service-specific-sdk}>=X.Y.Z

Main App Pattern

from fastapi import FastAPI
from api.routes import router
from core.config import settings

app = FastAPI(title="{Service} Vein", version="0.1.0")
app.include_router(router)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=settings.api_port)

Pattern: Simple FastAPI app, routes included at root or with prefix.

Testing Isolation

Because core/ is isolated from FastAPI:

# Can test core directly without HTTP
from vein.google.core.sheets import GoogleSheetsClient

def test_sheets_client():
    client = GoogleSheetsClient(mock_credentials)
    data = client.get_sheet_values("sheet_id", "A1:D10")
    assert len(data) > 0

Pattern: Core modules are testable without spinning up FastAPI.

Port Allocation

• 8001: Jira
• 8002: Slack
• 8003: Google
• 8004: WhatsApp (planned)
• 8005+: Future veins

Pattern: Sequential ports starting from 8001.

Vein Types

Type 1: Simple Token Auth

Examples: Jira, Slack, WhatsApp Auth: Token in headers or .env Stateless: No storage needed Inherits: BaseVein

Type 2: OAuth2

Examples: Google, GitHub, GitLab Auth: OAuth2 flow with callback Stateful: Requires token storage Inherits: BaseOAuthVein

Type 3: Hybrid (Future)

Examples: Services with webhooks + API May need: Database, Redis, webhook endpoints Consider: Pulse instead of vein (composed service)

When to Use Pulse vs Vein

Vein: Pure connector

• Stateless or minimal state (OAuth tokens)
• Pull-based (you call the API)
• No database required

Pulse: Composed service

• Stateful (database, message queue)
• Push-based (webhooks, real-time)
• Combines vein + storage + processing

Example: WhatsApp webhook receiver = pulse, WhatsApp API client = vein.

Standardization Checklist

When creating a new vein:

• Follow directory structure (core/, api/, models/)
• Create .env.example with all settings
• Implement /health endpoint
• Support ?text=true for all data endpoints
• Use from_{service}() factory methods in models
• Create text formatters in models/formatter.py
• Include README.md with setup instructions
• Choose correct base class (BaseVein or BaseOAuthVein)
• Allocate unique port (8001+)
• Keep core/ isolated from FastAPI

Evolution

This document captures patterns as of having 3 veins (Jira, Slack, Google).

Do not enforce these patterns rigidly - they should evolve as we build more veins. Do use this as a starting point for consistency. Do update this document when patterns change.

The abstract classes exist to enforce interfaces, not implementations. The patterns exist to reduce cognitive load, not to restrict flexibility.