# @psyche/database

Database library for Psyche Python services, aligned with `@oshun/database`.

## Features

- **PostgreSQL Client** - Async connection pooling with asyncpg
- **Redis Client** - Async Redis with cluster support
- **Transaction Management** - Retries, savepoints, advisory locks
- **Migration Runner** - Schema migrations with batch tracking
- **Health Checking** - Comprehensive health checks with caching
- **Query Builder** - Safe SQL construction with parameterized queries
- **SQLAlchemy Models** - ORM models for Psyche entities (with sqlalchemy extra)
- **Vector Storage** - Qdrant integration with embedding generation and
  similarity search

## Installation

```bash
# Core package
poetry add psyche-database

# With all optional dependencies
poetry add "psyche-database[all]"

# Specific extras
poetry add "psyche-database[redis]"      # Redis support
poetry add "psyche-database[sqlalchemy]" # SQLAlchemy integration
poetry add "psyche-database[migrations]" # Alembic migrations
poetry add "psyche-database[metrics]"    # Prometheus metrics
poetry add "psyche-database[vector]"     # Qdrant + OpenAI embeddings
poetry add "psyche-database[vector-all]" # Qdrant + OpenAI + Cohere
```

## Quick Start

### PostgreSQL Client

```python
from psyche_database import (
    PostgresConfig,
    create_postgres_client,
    QueryResult,
)

# Create client
config = PostgresConfig(
    host="localhost",
    port=5432,
    database="mydb",
    user="postgres",
    password="secret",
    min_connections=2,
    max_connections=10,
)

client = await create_postgres_client(config)

# Execute queries
result: QueryResult = await client.query(
    "SELECT * FROM users WHERE status = $1",
    ["active"]
)
for row in result.rows:
    print(row["email"])

# Get single row
user = await client.query_one(
    "SELECT * FROM users WHERE id = $1",
    [123]
)

# Execute statement
count = await client.execute(
    "UPDATE users SET status = $1 WHERE last_login < $2",
    ["inactive", "2024-01-01"]
)
print(f"Updated {count} rows")

# Close when done
await client.close()
```

### Transactions

```python
from psyche_database import (
    with_transaction,
    with_serializable_transaction,
    with_advisory_lock,
    RetryConfig,
)

# Basic transaction
async def create_order(conn):
    await conn.execute(
        "INSERT INTO orders (user_id, total) VALUES ($1, $2)",
        user_id, total
    )
    await conn.execute(
        "UPDATE inventory SET quantity = quantity - $1 WHERE product_id = $2",
        quantity, product_id
    )
    return order_id

result = await client.with_transaction(create_order)

# With retry on serialization failures
result = await with_transaction(
    client,
    create_order,
    retry=RetryConfig(max_retries=3)
)

# Serializable isolation level
result = await with_serializable_transaction(client, create_order)

# With advisory lock (for distributed locking)
async def process_user(conn):
    # Only one process can run this for user_id at a time
    return await update_user_stats(conn, user_id)

result = await with_advisory_lock(client, lock_id=user_id, fn=process_user)
```

### Redis Client

```python
from psyche_database import (
    RedisConfig,
    create_redis_client,
)

# Create client
config = RedisConfig(
    host="localhost",
    port=6379,
    password="secret",
    db=0,
    key_prefix="myapp:",
)

redis = await create_redis_client(config)

# Basic operations
await redis.set("user:123:name", "John")
name = await redis.get("user:123:name")

# With expiration
await redis.set("session:abc", "data", ex=3600)

# Hash operations
await redis.hset("user:123", mapping={"name": "John", "email": "john@example.com"})
user = await redis.hgetall("user:123")

# List operations
await redis.lpush("queue:tasks", "task1", "task2")
task = await redis.rpop("queue:tasks")

# Set operations
await redis.sadd("user:123:roles", "admin", "editor")
is_admin = await redis.sismember("user:123:roles", "admin")

await redis.close()
```

### Query Builder

```python
from psyche_database import (
    sql,
    sql_template,
    build_select_statement,
    build_insert_statement,
    build_where_clause,
    ComparisonOperator,
    WhereConditionWithOperator,
    OrderByClause,
    SortDirection,
    PaginationOptions,
)

# Simple parameterized query
query = sql("SELECT * FROM users WHERE id = $1", 123)
result = await client.query(query.text, query.values)

# Named template
query = sql_template(
    "SELECT * FROM users WHERE status = :status AND created_at > :since",
    status="active",
    since="2024-01-01"
)

# Build SELECT statement
query = build_select_statement(
    table="users",
    columns=["id", "name", "email"],
    where={"status": "active", "role": ["admin", "editor"]},
    order_by=OrderByClause(column="created_at", direction=SortDirection.DESC),
    pagination=PaginationOptions(limit=10, offset=20),
)

# Build INSERT with ON CONFLICT
query = build_insert_statement(
    table="users",
    data={"email": "john@example.com", "name": "John"},
    returning=["id"],
    on_conflict_columns=["email"],
    on_conflict_action=["name"],  # Update name on conflict
)

# Complex WHERE conditions
conditions = [
    WhereConditionWithOperator(
        column="age",
        operator=ComparisonOperator.GE,
        value=18
    ),
    WhereConditionWithOperator(
        column="status",
        operator=ComparisonOperator.IN,
        values=["active", "pending"]
    ),
]
where = build_where_clause(conditions)
```

### Migrations

```python
from psyche_database import (
    create_migration_runner,
    create_sql_migration,
    Migration,
    MigrationConfig,
)

# Create migration runner
runner = create_migration_runner(
    client,
    MigrationConfig(table_name="migrations", schema_name="public")
)

# Register SQL migrations
runner.register_many([
    create_sql_migration(
        "20240101120000_create_users",
        "Create users table",
        up_sql="""
            CREATE TABLE users (
                id SERIAL PRIMARY KEY,
                email VARCHAR(255) UNIQUE NOT NULL,
                created_at TIMESTAMPTZ DEFAULT NOW()
            )
        """,
        down_sql="DROP TABLE users"
    ),
    create_sql_migration(
        "20240102120000_add_user_name",
        "Add name to users",
        up_sql="ALTER TABLE users ADD COLUMN name VARCHAR(255)",
        down_sql="ALTER TABLE users DROP COLUMN name"
    ),
])

# Run migrations
summary = await runner.up()
print(f"Applied: {summary.applied}")

# Check status
status = await runner.status()
print(f"Applied: {len(status.applied)}, Pending: {len(status.pending)}")

# Rollback last batch
summary = await runner.down()
```

### Health Checks

```python
from psyche_database import (
    check_all_databases_health,
    create_health_monitor,
    create_health_response,
    get_health_status_code,
    HealthCheckConfig,
)

# One-time health check
health = await check_all_databases_health(
    postgres=pg_client,
    redis=redis_client,
)
print(f"Overall: {health.overall}")

# FastAPI integration
from fastapi import FastAPI, Response

app = FastAPI()

@app.get("/health")
async def health_check(response: Response):
    health = await check_all_databases_health(
        postgres=pg_client,
        redis=redis_client,
    )
    response.status_code = get_health_status_code(health)
    return create_health_response(health)

# Continuous monitoring
def on_health_change(health):
    if health.overall != "healthy":
        logger.warning(f"Database health: {health.overall}")

monitor = create_health_monitor(
    postgres=pg_client,
    redis=redis_client,
    interval_ms=30000,
    callback=on_health_change,
)
monitor.start()
# ... later ...
monitor.stop()
```

### Connection Strings

```python
from psyche_database import (
    parse_postgres_connection_string,
    build_postgres_connection_string,
    mask_postgres_connection_string,
    detect_database_type,
    validate_connection_string,
    create_postgres_client_from_url,
)

# Parse connection string
config = parse_postgres_connection_string(
    "postgresql://user:pass@localhost:5432/mydb?sslmode=require"
)
print(f"Host: {config.host}, Database: {config.database}")

# Build connection string
url = build_postgres_connection_string(
    PostgresConnectionStringOptions(
        host="localhost",
        database="mydb",
        user="user",
        password="secret",
        ssl=True,
    )
)

# Mask for logging
masked = mask_postgres_connection_string(url)
print(masked)  # postgresql://user:****@localhost:5432/mydb

# Validate
result = validate_connection_string(url)
if not result.valid:
    print(f"Invalid: {result.error}")

# Create client from URL
client = await create_postgres_client_from_url(
    "postgresql://user:pass@localhost:5432/mydb"
)
```

## Configuration

### Environment Variables

```bash
# PostgreSQL
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
POSTGRES_DATABASE=mydb
POSTGRES_USER=postgres
POSTGRES_PASSWORD=secret
POSTGRES_SSL=true
POSTGRES_MIN_CONNECTIONS=2
POSTGRES_MAX_CONNECTIONS=10

# Redis
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_PASSWORD=secret
REDIS_DB=0
REDIS_TLS=false
REDIS_KEY_PREFIX=myapp:

# Connection URLs
DATABASE_URL=postgresql://user:pass@localhost:5432/mydb
REDIS_URL=redis://localhost:6379/0
```

### From Environment

```python
from psyche_database import (
    create_postgres_client_from_env,
    create_redis_client_from_env,
)

# Uses POSTGRES_* environment variables
pg_client = await create_postgres_client_from_env()

# Uses REDIS_* environment variables
redis_client = await create_redis_client_from_env()
```

## SQLAlchemy Models

When installed with the `sqlalchemy` extra, the library provides ORM models for
core Psyche entities.

### Available Models

```python
from psyche_database import (
    # Base classes
    Base,
    TimestampMixin,
    UUIDMixin,

    # Persona models
    Persona,
    PersonaVersion,
    PersonaTemplate,
    PersonaRole,
    PersonaStatus,

    # Session models
    Session,
    SessionState,
    SessionStats,

    # Memory models (with pgvector support)
    Memory,
    MemoryType,
    MemoryScope,

    # Conversation models
    Conversation,
    Message,
    MessageRole,

    # Tool execution models
    ToolExecution,
    ToolExecutionStatus,
)
```

### Persona Example

```python
from psyche_database import Persona, PersonaRole, PersonaStatus

# Create a persona
persona = Persona(
    name="customer-support",
    display_name="Support Assistant",
    slug="customer-support",
    role=PersonaRole.ASSISTANT.value,
    status=PersonaStatus.ACTIVE.value,
    personality={
        "big_five": {
            "openness": 0.7,
            "conscientiousness": 0.9,
            "extraversion": 0.6,
            "agreeableness": 0.8,
            "neuroticism": 0.2,
        },
        "communication": {
            "style": "professional",
            "formality_level": 0.7,
            "humor_level": 0.3,
        },
    },
    expertise={
        "domains": ["customer-service", "product-support"],
        "primary_domain": "customer-service",
    },
)
```

### Memory with Embeddings

```python
from psyche_database import Memory, MemoryType, MemoryScope

# Store a memory with embedding
memory = Memory(
    user_id=user_id,
    persona_id=persona_id,
    memory_type=MemoryType.PREFERENCE.value,
    scope=MemoryScope.USER.value,
    content="User prefers concise responses",
    embedding=[0.1, 0.2, ...],  # 1536-dim vector
    embedding_model="text-embedding-ada-002",
    importance=0.8,
)
```

### Session Tracking

```python
from psyche_database import Session, SessionState

# Create and manage a session
session = Session(
    user_id=user_id,
    persona_id=persona_id,
    channel="web",
)
session.start()

# Update activity
session.touch()

# End session with summary
session.end(summary="Helped user with billing inquiry")
```

## Vector Storage

When installed with the `vector` extra, the library provides Qdrant integration,
embedding generation, and similarity search utilities.

### Installation

```bash
# With OpenAI embeddings (recommended)
poetry add "psyche-database[vector]"

# With OpenAI and Cohere embeddings
poetry add "psyche-database[vector-all]"
```

### Configuration

```bash
# Qdrant
QDRANT_HOST=localhost
QDRANT_PORT=6333
QDRANT_API_KEY=your-api-key  # Optional

# OpenAI embeddings
OPENAI_API_KEY=your-openai-key

# Cohere embeddings (optional)
COHERE_API_KEY=your-cohere-key
```

### Embedding Generation

```python
from psyche_database import (
    create_embedding_service,
    EmbeddingProvider,
    get_embedding_dimensions,
)

# Create OpenAI embedding service
embeddings = create_embedding_service(
    provider=EmbeddingProvider.OPENAI,
    model="text-embedding-3-small",
)

# Generate embeddings
texts = ["Hello world", "How are you?"]
vectors = await embeddings.embed(texts)

# Single text
vector = await embeddings.embed_single("Hello world")

# Get dimensions for a model
dims = get_embedding_dimensions("text-embedding-3-small")  # 1536
```

### Qdrant Vector Store

```python
from psyche_database import (
    create_qdrant_client_from_env,
    CollectionConfig,
    VectorDocument,
    SearchOptions,
)

# Create client from environment
store = create_qdrant_client_from_env()

# Create a collection
await store.create_collection(
    CollectionConfig(
        name="documents",
        vector_size=1536,
        distance="Cosine",
    )
)

# Upsert documents
docs = [
    VectorDocument(
        id="doc-1",
        content="Hello world",
        embedding=[0.1, 0.2, ...],  # 1536-dim vector
        metadata={"source": "web"},
    ),
]
count = await store.upsert("documents", docs)

# Search by vector
results = await store.search(
    "documents",
    query_vector=[0.1, 0.2, ...],
    options=SearchOptions(
        limit=10,
        score_threshold=0.7,
        filter={"source": "web"},
    ),
)

# Retrieve by IDs
docs = await store.get("documents", ["doc-1", "doc-2"])

# Delete documents
await store.delete("documents", ["doc-1"])

# Collection info
info = await store.get_collection_info("documents")
print(f"Vectors: {info.vector_count}")
```

### Similarity Search

High-level search interface combining embeddings and vector store:

```python
from psyche_database import (
    create_similarity_search,
    SimilaritySearchConfig,
    VectorDocument,
)

# Create similarity search instance
search = create_similarity_search(
    collection_name="documents",
    embedding_provider="openai",
    embedding_model="text-embedding-3-small",
)

# Index documents (generates embeddings automatically)
docs = [
    VectorDocument(id="doc-1", content="Python is a programming language"),
    VectorDocument(id="doc-2", content="JavaScript runs in browsers"),
    VectorDocument(id="doc-3", content="Rust is a systems language"),
]
await search.index(docs)

# Search by text (converts to embedding automatically)
results = await search.search(
    query="best language for web development",
    limit=5,
    score_threshold=0.5,
)

for result in results:
    print(f"{result.id}: {result.score:.3f} - {result.content}")

# Batch search
queries = ["web development", "systems programming"]
batch_results = await search.search_batch(queries, limit=3)
```

### Hybrid Search

Combines vector similarity with keyword matching using Reciprocal Rank Fusion:

```python
from psyche_database import (
    HybridSearch,
    HybridSearchConfig,
    create_qdrant_client_from_env,
    create_embedding_service,
)

# Create hybrid search
config = HybridSearchConfig(
    collection_name="documents",
    vector_weight=0.7,
    keyword_weight=0.3,
)

store = create_qdrant_client_from_env()
embeddings = create_embedding_service()

hybrid = HybridSearch(store, embeddings, config)

# Search with hybrid ranking
results = await hybrid.search(
    query="Python web framework",
    limit=10,
)
```

### Caching

SimilaritySearch includes built-in caching to avoid redundant embedding
generation:

```python
from psyche_database import create_similarity_search

search = create_similarity_search(
    collection_name="documents",
    enable_cache=True,       # Enable caching (default)
    cache_ttl_seconds=3600,  # 1 hour TTL
)

# First search generates embedding
results = await search.search("hello world")

# Second identical search uses cached embedding
results = await search.search("hello world")

# Clear cache manually
search.clear_cache()
```

## Alignment with @oshun/database

This library provides Python equivalents for all key types and functions:

| TypeScript                | Python                       |
| ------------------------- | ---------------------------- |
| `PostgresConfig`          | `PostgresConfig`             |
| `PostgresClient`          | `PostgresClient`             |
| `RedisClient`             | `RedisClient`                |
| `RedisClusterClient`      | `RedisClusterClient`         |
| `TransactionOptions`      | `TransactionOptions`         |
| `withTransaction()`       | `with_transaction()`         |
| `MigrationRunner`         | `MigrationRunner`            |
| `sql\`...\``              | `sql()`, `sql_template()`    |
| `buildWhereClause()`      | `build_where_clause()`       |
| `checkAllDatabasesHealth` | `check_all_databases_health` |
| `HealthMonitor`           | `HealthMonitor`              |
| `QdrantVectorStore`       | `QdrantVectorStore`          |
| `EmbeddingService`        | `EmbeddingService`           |
| `SimilaritySearch`        | `SimilaritySearch`           |
| `HybridSearch`            | `HybridSearch`               |

## Database Migrations

The library includes Alembic migrations for schema management.

### Installation

```bash
poetry add "psyche-database[migrations]"
```

### Quick Start

```bash
# Set database URL
export DATABASE_URL="postgresql://user:pass@localhost:5432/psyche"

# Apply all migrations
alembic -c $(python -c "from psyche_database.migrations import ALEMBIC_INI; print(ALEMBIC_INI)") upgrade head

# Check current version
alembic current

# Show migration history
alembic history --verbose
```

### Available Migrations

| Revision     | Description                                     |
| ------------ | ----------------------------------------------- |
| 001_initial  | Core schema (personas, sessions, memories, etc) |
| 002_pgvector | pgvector support for semantic search            |

### Generating New Migrations

```bash
# Auto-generate from model changes
alembic revision --autogenerate -m "add new column"

# Create empty migration
alembic revision -m "custom migration"
```

### Migration Strategy

For detailed migration strategy from Serwaa to Psyche, see
[MIGRATION_STRATEGY.md](./MIGRATION_STRATEGY.md).

## Nx Targets

| Target         | Description             |
| -------------- | ----------------------- |
| `build`        | Build the package       |
| `install`      | Install dependencies    |
| `lint`         | Run Ruff linter         |
| `lint-fix`     | Fix linting issues      |
| `format`       | Format with Black       |
| `format-check` | Check formatting        |
| `typecheck`    | Run MyPy                |
| `test`         | Run pytest              |
| `test-cov`     | Run tests with coverage |

## License

Proprietary - Oshun Platform