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

```python
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)

```python
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)

```python
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

```python
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

```python
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

```python
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

```python
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**:

```python
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

```python
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:

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

# Vein config
API_PORT=8001
```

### requirements.txt

Minimal dependencies:

```txt
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

```python
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:

```python
# 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.