Compare commits

...
21 Commits
Author SHA1 Message Date
Jeroen SchweitzerandClaude Opus 4.5 ffa984e271 fix: separate httpx and SQLAlchemy async contexts in bulk sync
Build and Push / build (release) Successful in 50s
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-01 21:34:34 +01:00
Jeroen SchweitzerandClaude Opus 4.5 7d13be6052 fix: use uuid field instead of pk for Authentik user sync
Build and Push / build (release) Successful in 50s
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-01 21:27:21 +01:00
Jeroen SchweitzerandClaude Opus 4.5 faff45db90 fix: manually handle session cookies for Authentik API authentication
Build and Push / build (release) Successful in 1m14s
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-01 21:01:29 +01:00
Jeroen SchweitzerandClaude Opus 4.5 892015fa59 fix: use Authentik domain URL for cookie domain compatibility
Build and Push / build (release) Successful in 1m17s
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-01 20:52:14 +01:00
Jeroen SchweitzerandClaude Opus 4.5 ee93a73160 fix: use AUTHENTIK_USERNAME/PASSWORD env vars to match production
Build and Push / build (release) Successful in 28s
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-01 20:37:12 +01:00
Jeroen SchweitzerandClaude Opus 4.5 4f45f9bf37 feat: add authentication and user management with Authentik integration
Build and Push / build (release) Successful in 1m16s
- Add PostgreSQL database with async SQLAlchemy
- Add Alembic migrations for schema management
- Add User, Role, UserPreferences, ApiKey models
- Add auth endpoints: /auth/me, /auth/users, /auth/users/sync-from-authentik
- Add token validation via Authentik userinfo endpoint
- Add bulk user sync from Authentik admin API
- Add database health check to diagnostics

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-01 20:25:03 +01:00
jpmschweitzerandClaude Opus 4.5 cdb6344013 feat: add PostgreSQL database configuration
- Add POSTGRES_HOST, POSTGRES_USER, POSTGRES_PASSWORD settings
- Add database_url property for connection string construction
- Database: core_api on shared postgres instance

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-01 15:43:44 +01:00
jpmschweitzerandClaude Opus 4.5 0d2926dc6b fix: remove hardcoded service URLs, require ENV config
Build and Push / build (release) Successful in 29s
- Remove hardcoded default URLs (portainer, npm, ollama, etc.)
- All external service URLs now required via ENV vars
- Remove obsolete ai_client and ai_controller (core-ai proxy)
- Clean up health_controller obsolete core-ai references
- App fails fast at startup if required config missing

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-31 11:59:46 +01:00
jpmschweitzerandClaude Opus 4.5 31370f053c fix: clean up root endpoint and disable redoc
Build and Push / build (release) Successful in 28s
- Simplify root endpoint to return minimal status info
- Remove obsolete endpoint listings (moved to /docs)
- Clean up verbose app description
- Disable redoc UI

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-31 11:46:26 +01:00
jpmschweitzerandClaude Opus 4.5 9370c70311 fix: correct Portainer default port to 8001
Port 9000 is Authentik, Portainer runs on 8001

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-31 11:29:18 +01:00
jpmschweitzerandClaude Opus 4.5 68909e6576 feat: add development server startup script
Build and Push / build (release) Successful in 29s
Add wakeup.sh for local development with:
- Port conflict detection
- Virtual environment activation
- Log file management
- Colored output

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-31 11:15:17 +01:00
jpmschweitzerandClaude Opus 4.5 435a3faeed chore: clean up .gitignore
Remove duplicate credentials.py entry

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-31 11:15:01 +01:00
jpmschweitzerandClaude Opus 4.5 06cb4c17c8 chore: bump version to 1.3.1
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-31 11:14:27 +01:00
jpmschweitzerandClaude Opus 4.5 ffa075528b test: update tests for Portainer client without socket fallback
- Remove TestPortainerClientDockerSocketFallback class
- Update wrapper method tests to expect RuntimeError on missing endpoints
- Remove all socket fallback related test cases

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-31 11:14:04 +01:00
jpmschweitzerandClaude Opus 4.5 dbbb92d292 refactor: remove Docker socket fallback from Portainer client
- Remove _list_containers_via_socket method
- Remove _inspect_container_via_socket method
- Simplify list_containers and inspect_container to use Portainer API only
- Raise RuntimeError when no Portainer endpoints available

BREAKING: Portainer API configuration is now required for all container
and stack operations. Set PORTAINER_URL and PORTAINER_API_KEY env vars.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-31 11:13:46 +01:00
jpmschweitzerandClaude Opus 4.5 3ecfb91cc2 refactor: simplify configuration to use only environment variables
- Remove credentials.py import, use pydantic-settings .env support
- Remove unused search API keys (Brave, Google)
- Update default search provider to SearXNG
- Add extra="ignore" to allow flexible env var usage
- Update .env.example with organized sections

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-31 11:13:26 +01:00
jpmschweitzerandClaude Opus 4.5 167a3a43a8 feat: add stack management and container delete endpoints
Build and Push / build (release) Successful in 29s
- Add stack compose YAML get/update endpoints
- Add stack environment variable get/update endpoints
- Add stack deploy and rebuild endpoints
- Add container delete endpoint with force option
- Add Portainer client methods for new operations
- Add comprehensive test coverage (30 new tests)
- Remove REQUESTED_SERVICES.md (now implemented)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-31 10:19:50 +01:00
jpmschweitzerandClaude Opus 4.5 b5bdff3d54 Bump version to 1.2.1
Build and Push / build (release) Successful in 28s
🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-17 18:51:00 +01:00
jpmschweitzerandClaude Opus 4.5 5c4ac70450 Add AGENTS.md with project guidelines
Documents coding standards, git discipline, release flow,
and FastAPI architecture best practices.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-17 18:46:15 +01:00
jpmschweitzerandClaude Opus 4.5 ad1b55009a Remove obsolete web scraper config from .env.example
Remove APP_VERSION (now from pyproject.toml) and web scraper settings.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-17 18:46:02 +01:00
jpmschweitzerandClaude Opus 4.5 fd8aee3227 Fix device control response returning stale state
Add 300ms delay after executing device action before fetching new state,
allowing Home Assistant time to update the entity state.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-17 18:40:19 +01:00
35 changed files with 3200 additions and 1304 deletions
+54 -12
View File
@@ -1,23 +1,65 @@
# Core Code API Configuration # Core Code API Configuration
# Copy to .env and fill in real values
# Application settings # =============================================================================
# Application
# =============================================================================
APP_NAME="Core Code API" APP_NAME="Core Code API"
APP_VERSION="1.0.0"
DEBUG=false DEBUG=false
LOG_LEVEL=INFO
# Server settings # =============================================================================
# Server
# =============================================================================
HOST=0.0.0.0 HOST=0.0.0.0
PORT=8083 PORT=8083
# CORS settings (default allows all origins for internal use) # CORS (default allows all origins for internal use)
# CORS_ORIGINS=["http://192.168.86.149:82"] # CORS_ORIGINS=["http://192.168.86.149:82"]
# Logging # =============================================================================
LOG_LEVEL=INFO # Infrastructure Services
# =============================================================================
# Web Scraper Module # Portainer API (required for container/stack management)
WEB_SCRAPER_REQUEST_TIMEOUT=30 PORTAINER_URL=http://localhost:8001
WEB_SCRAPER_MAX_REDIRECTS=5 PORTAINER_API_KEY=ptr_your-api-key-here
WEB_SCRAPER_USER_AGENT="Mozilla/5.0 (compatible; CoreCode/1.0)"
WEB_SCRAPER_DEFAULT_MAX_LENGTH=10000 # Nginx Proxy Manager API
WEB_SCRAPER_MAX_LINKS_TO_EXTRACT=50 NPM_URL=http://localhost:81
NPM_EMAIL=admin@example.com
NPM_PASSWORD=your-npm-password
# =============================================================================
# Home Automation
# =============================================================================
# Home Assistant API
HOMEASSISTANT_URL=http://localhost:8123
HOMEASSISTANT_TOKEN=your-long-lived-access-token
# =============================================================================
# AI Services
# =============================================================================
# Ollama API
OLLAMA_BASE_URL=http://localhost:11434
# SearXNG (self-hosted search)
SEARXNG_URL=http://localhost:8080
# =============================================================================
# Database (PostgreSQL)
# =============================================================================
POSTGRES_HOST=localhost:5432
POSTGRES_USER=core_api
POSTGRES_PASSWORD=your-password
# =============================================================================
# Vector Database
# =============================================================================
# Qdrant
QDRANT_HOST=qdrant
QDRANT_PORT=6333
-1
View File
@@ -105,7 +105,6 @@ data/
# Credentials and secrets # Credentials and secrets
src/credentials.py src/credentials.py
credentials.py
*.pem *.pem
*.key *.key
secrets/ secrets/
+72
View File
@@ -0,0 +1,72 @@
# AGENTS.md
> **Start every session by reading this file.**
> This file outlines the operational protocols, coding standards, and architectural decisions for this FastAPI project.
## 1. Agent Operational Protocols
### 🧠 Work Patterns (Plan-Act-Reflect)
* **Plan:** Before writing code, briefly outline your plan. Identify which files you will touch and what the side effects might be.
* **Act:** Execute the changes in small, atomic steps.
* **Reflect:** After coding, verify your work. Did you break existing tests? Did you add new tests?
### 🛡️ Git Discipline
* **NEVER commit to `main` or `master` directly.** Always create a feature branch: `feature/your-feature-name` or `fix/issue-description`.
* **Commit Messages:** Use the [Conventional Commits](https://www.conventionalcommits.org/) format.
* `feat: add user login endpoint`
* `fix: resolve database connection timeout`
* `refactor: split monolith dependency file`
* **Atomic Commits:** Keep commits small. One logical change = one commit.
### 📝 Changelog Maintenance
* **Update `CHANGELOG.md`** with every user-facing change.
* Format: `## [Unreleased] - YYYY-MM-DD` followed by `### Added`, `### Changed`, or `### Fixed`.
### 🚀 Release Flow
When changes are ready for deployment:
1. **Ask user if deploy cycle is desired **
2. **Update version** in `pyproject.toml`:
- Bug fixes: bump patch version (1.8.3 → 1.8.4)
- New features: bump minor version (1.8.4 → 1.9.0)
3. **Update CHANGELOG.md**:
- Move items from `[Unreleased]` to new version section
- Add release date: `## [1.8.4] - 2025-12-16`
4. **Commit and tag**:
```bash
git add -A
git commit -m "fix: description of changes"
git tag v1.8.4
git push origin main --tags
```
5. **CI/CD triggers automatically**:
- Gitea CI builds Docker image on new tag
- Watchtower pulls and deploys to production
- Verify deployment: `curl http://192.168.86.149:8000/health`
---
## 2. FastAPI Architecture & Best Practices
*Reference: [FastAPI Best Practices](https://github.com/zhanymkanov/fastapi-best-practices)*
### 📂 Project Structure (Directory-based, NOT File-type based)
Do **not** group files by type (e.g., one huge `routers` folder). Group by **domain/module** inside a `src/` directory.
**Correct Structure:**
```text
src/
├── auth/
│ ├── router.py # Endpoints
│ ├── schemas.py # Pydantic models
│ ├── service.py # Business logic (CRUD, etc.)
│ ├── dependencies.py# Module-specific dependencies
│ └── config.py # Module-specific settings
├── posts/
│ ├── router.py
│ └── ...
└── main.py # App entry point
+97
View File
@@ -5,6 +5,103 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [1.4.5] - 2026-01-01
### Fixed
- Separate httpx and SQLAlchemy async contexts in bulk sync (fixes greenlet error)
## [1.4.4] - 2026-01-01
### Fixed
- Use `uuid` field instead of `pk` for Authentik user sync (pk is integer, uuid is proper UUID)
- Skip internal_service_account type users during bulk sync
## [1.4.3] - 2026-01-01
### Fixed
- Manually extract and send session cookies for Authentik flow auth (fixes cross-domain cookie handling)
## [1.4.2] - 2026-01-01
### Fixed
- Use Authentik domain URL (auth.schweitz.net) instead of IP to fix cookie domain matching
## [1.4.1] - 2026-01-01
### Fixed
- Authentik config now uses AUTHENTIK_USERNAME/PASSWORD to match production env vars
## [1.4.0] - 2026-01-01
### Added
- **Authentication & User Management** - Authentik integration for user synchronization
- `GET /auth/me` - Get current authenticated user info
- `GET /auth/users` - List all users with search and pagination
- `POST /auth/users/sync-from-authentik` - Bulk sync users from Authentik admin API
- PostgreSQL database integration with async SQLAlchemy
- Alembic database migrations for schema management
- Database models: User, Role, UserPreferences, ApiKey
- Token validation via Authentik userinfo endpoint
- Role synchronization from Authentik groups
- Health check now includes database connectivity status
### Changed
- Authentik configuration now uses username/password for admin API access
- Health endpoint includes database status in diagnostics
## [1.3.1] - 2025-12-31
### Changed
- Simplified configuration: all settings now read from environment variables/.env only
- Removed Docker socket fallback for container operations (Portainer API is now required)
- Updated default search provider to SearXNG
### Removed
- `src/credentials.py` - credentials now managed via environment variables
- Docker socket fallback methods from Portainer client
- Unused search API settings (Brave, Google)
## [1.3.0] - 2025-12-31
### Added
- **Stack Management Endpoints** for Tatlock Control Room integration
- `GET /infrastructure/stacks/{stackId}/compose` - Get stack Docker Compose YAML
- `PUT /infrastructure/stacks/{stackId}/compose` - Update stack Docker Compose YAML
- `GET /infrastructure/stacks/{stackId}/env` - Get stack environment variables
- `PUT /infrastructure/stacks/{stackId}/env` - Update stack environment variables
- `POST /infrastructure/stacks/{stackId}/deploy` - Redeploy stack
- `POST /infrastructure/stacks/{stackId}/rebuild` - Pull images and recreate containers
- `DELETE /infrastructure/containers/{id}` - Delete container with optional force flag
- Portainer client methods: `get_stack_file`, `redeploy_stack`, `update_stack_env`, `delete_container`, `restart_container`
### Removed
- REQUESTED_SERVICES.md - endpoint specifications now implemented
## [1.2.1] - 2025-12-17
### Fixed
- Device control endpoint now returns correct new state after action (added 300ms delay for HA state propagation)
### Added
- AGENTS.md with project coding guidelines and release flow documentation
### Changed
- Removed obsolete web scraper settings from .env.example
## [1.2.0] - 2025-12-17 ## [1.2.0] - 2025-12-17
### Added ### Added
-446
View File
@@ -1,446 +0,0 @@
# Requested Core-API Services for Core-AI Infrastructure Tools
This document specifies the API endpoints needed by core-ai infrastructure tools. All requests from core-ai should go through core-api for centralized logging and access control.
## Context
The core-ai service is implementing 9 infrastructure tools in 3 logical clusters:
1. **Container Lifecycle** (4 tools) - containers.py
2. **Service Management** (3 tools) - services.py
3. **Monitoring & Resources** (2 tools) - monitoring.py
These tools need corresponding core-api REST endpoints to perform operations via Portainer.
---
## Cluster 1: Container Lifecycle Management
### 1.1 List Containers
**Endpoint:** `GET /v1/infrastructure/containers`
**Query Parameters:**
- `status` (optional): Filter by status - "all", "running", "stopped", "paused" (default: "running")
**Response:**
```json
[
{
"Id": "abc123...",
"Names": ["/nginx"],
"State": "running",
"Status": "Up 3 days",
"Image": "nginx:latest",
"Ports": [
{"PrivatePort": 80, "PublicPort": 8080, "Type": "tcp"},
{"PrivatePort": 443, "PublicPort": 8443, "Type": "tcp"}
],
"StartedAt": "2024-12-01T10:00:00Z"
}
]
```
**Implementation Notes:**
- Use `PortainerClient.list_containers(all_containers=True)` with Docker socket fallback
- Filter results based on `status` query parameter
- Return standard Docker API container list format
---
### 1.2 Manage Container
**Endpoint:** `POST /v1/infrastructure/containers/{container}/{action}`
**Path Parameters:**
- `container`: Container name or ID (e.g., "nginx", "core-ai")
- `action`: One of: "start", "stop", "restart", "pause", "unpause", "remove"
**Response (Success):**
```json
{
"success": true,
"action": "restart",
"container": "nginx",
"message": "Container restarted successfully"
}
```
**Response (Error):**
```json
{
"success": false,
"error": "Container not found",
"message": "Container 'nginx2' not found. Available containers: nginx, core-ai, ollama"
}
```
**Status Codes:**
- `200` - Success
- `304` - Not Modified (already in target state)
- `404` - Container not found
- `409` - Conflict (e.g., cannot remove running container)
- `500` - Server error
**Implementation Notes:**
- For "restart": call stop then start
- For actions not yet in PortainerClient (pause, unpause, remove):
- Call Portainer API directly: `/api/endpoints/{endpoint_id}/docker/containers/{container_id}/{action}`
- Handle partial name matching (case-insensitive)
- Return helpful error messages suggesting `docker_list_containers()` when not found
---
### 1.3 Inspect Container
**Endpoint:** `GET /v1/infrastructure/containers/{container}`
**Path Parameters:**
- `container`: Container name or ID
**Query Parameters:**
- `details` (optional): Level of detail - "summary" (default), "full", "resources"
**Response:**
```json
{
"Id": "abc123...",
"Name": "/nginx",
"State": {
"Status": "running",
"Running": true,
"StartedAt": "2024-12-01T10:00:00Z",
"FinishedAt": "0001-01-01T00:00:00Z",
"ExitCode": 0
},
"Config": {
"Image": "nginx:latest",
"Env": ["PATH=/usr/local/sbin:...", "NGINX_VERSION=1.25.0"],
"Cmd": ["nginx", "-g", "daemon off;"]
},
"NetworkSettings": {
"Ports": {
"80/tcp": [{"HostIp": "0.0.0.0", "HostPort": "8080"}],
"443/tcp": [{"HostIp": "0.0.0.0", "HostPort": "8443"}]
},
"Networks": {
"bridge": {
"IPAddress": "172.17.0.2",
"Gateway": "172.17.0.1"
}
}
},
"HostConfig": {
"Memory": 536870912,
"NanoCpus": 1000000000,
"RestartPolicy": {"Name": "unless-stopped"}
},
"Mounts": [
{
"Type": "bind",
"Source": "/host/path",
"Destination": "/container/path"
}
]
}
```
**Implementation Notes:**
- Use `PortainerClient.inspect_container(container)` which auto-detects endpoint and falls back to Docker socket
- Return full Docker inspect response
- The core-ai tool will handle formatting based on `details` level
- Return 404 if container not found
---
### 1.4 Container Logs
**Endpoint:** `GET /v1/infrastructure/containers/{container}/logs`
**Path Parameters:**
- `container`: Container name or ID
**Query Parameters:**
- `lines` (optional): Number of log lines (default: 50, max: 500)
- `since` (optional): Time filter - "1h", "30m", or ISO timestamp
**Response:**
```json
{
"container": "nginx",
"lines_requested": 50,
"since": null,
"logs": "2024-12-04T10:00:00.123Z Starting nginx...\n2024-12-04T10:00:01.456Z Ready to accept connections\n..."
}
```
**Implementation Notes:**
- Access Docker API directly: `GET /v1.41/containers/{container}/logs`
- Use Docker socket transport (httpx with uds)
- Parameters: `stdout=true`, `stderr=true`, `tail={lines}`, `timestamps=true`
- If `since` provided: add `since={unix_timestamp}` parameter
- Strip Docker stream headers (8-byte binary prefix per line)
- Return plain text logs with timestamps
- Return 404 if container not found
---
## Cluster 2: Service Management
### 2.1 List Services
**Endpoint:** `GET /v1/infrastructure/services` (already exists, may need enhancement)
**Query Parameters:**
- `stack` (optional): Filter by stack name
**Response:**
```json
[
{
"name": "portainer",
"stack_id": 1,
"status": "active",
"containers_running": 3,
"containers_total": 3,
"ports": [9000, 8000],
"domains": ["portainer.example.com"]
}
]
```
**Implementation Notes:**
- Enhance existing `/infrastructure/services` endpoint if needed
- Ensure it returns stack/service information from Portainer
- Include container counts (running/total)
---
### 2.2 Manage Service
**Endpoint:** `POST /v1/infrastructure/services/{service}/{action}`
**Path Parameters:**
- `service`: Service/stack name
- `action`: One of: "start", "stop", "restart", "scale"
**Request Body (for scale action):**
```json
{
"replicas": 3
}
```
**Response:**
```json
{
"success": true,
"action": "restart",
"service": "web",
"message": "Service restarted successfully"
}
```
**Implementation Notes:**
- For "start"/"stop": Use Portainer stack start/stop API
- For "restart": Stop then start the stack
- For "scale": Update stack with new replica count
- This may require updating stack compose file
---
### 2.3 Service Status
**Endpoint:** `GET /v1/infrastructure/services/{service}/status`
**Path Parameters:**
- `service`: Service/stack name
**Response:**
```json
{
"name": "web",
"status": "active",
"stack_id": 5,
"containers": [
{
"name": "web_app_1",
"status": "running",
"health": "healthy",
"uptime": "2 days"
}
],
"replica_status": "3/3 running",
"resources": {
"memory_total": "1.2 GB",
"cpu_usage": "15%"
},
"recent_events": [
{"time": "2024-12-04T09:00:00Z", "action": "container_start", "container": "web_app_3"}
]
}
```
**Implementation Notes:**
- Get stack details from Portainer
- Get individual container statuses
- Calculate aggregate resource usage
- May require querying Docker events API for recent events
---
## Cluster 3: Monitoring & Resources
### 3.1 System Resources
**Endpoint:** `GET /v1/infrastructure/resources/system`
**Response:**
```json
{
"cpu": {
"cores": 8,
"usage_percent": 45.2,
"load_average": [2.5, 2.3, 2.1]
},
"memory": {
"total_bytes": 16777216000,
"used_bytes": 8388608000,
"available_bytes": 8388608000,
"usage_percent": 50.0
},
"disk": {
"total_bytes": 500000000000,
"used_bytes": 250000000000,
"available_bytes": 250000000000,
"usage_percent": 50.0
},
"network": {
"interfaces": {
"eth0": {
"rx_bytes": 1000000000,
"tx_bytes": 500000000
}
}
}
}
```
**Implementation Notes:**
- Use Docker system info API: `GET /v1.41/system/df`
- May also use `GET /v1.41/info` for system-wide stats
- Calculate percentages and format nicely
- Include load averages from system stats
---
### 3.2 Container Resources
**Endpoint:** `GET /v1/infrastructure/resources/containers`
**Query Parameters:**
- `container` (optional): Specific container name/ID (if omitted, return all)
**Response:**
```json
[
{
"name": "nginx",
"cpu_percent": 5.2,
"memory_usage_bytes": 45000000,
"memory_limit_bytes": 100000000,
"memory_percent": 45.0,
"network_rx_bytes": 50000000,
"network_tx_bytes": 25000000,
"block_read_bytes": 10000000,
"block_write_bytes": 5000000
}
]
```
**Implementation Notes:**
- Use Docker stats API: `GET /v1.41/containers/{id}/stats?stream=false`
- If `container` param provided: return single container stats
- If omitted: return stats for all running containers
- Calculate percentages where applicable
- Stats API returns real-time metrics (one-time snapshot, not streaming)
---
## Implementation Priority
**Phase 1 (Needed immediately for core-ai):**
1. `GET /v1/infrastructure/containers` - List containers
2. `POST /v1/infrastructure/containers/{container}/{action}` - Manage containers
3. `GET /v1/infrastructure/containers/{container}` - Inspect container
4. `GET /v1/infrastructure/containers/{container}/logs` - Container logs
**Phase 2 (Needed for full infrastructure tools):**
5. `POST /v1/infrastructure/services/{service}/{action}` - Manage services
6. `GET /v1/infrastructure/services/{service}/status` - Service status
7. `GET /v1/infrastructure/resources/system` - System resources
8. `GET /v1/infrastructure/resources/containers` - Container resources
---
## Security & Access Control
All endpoints should:
- Log all requests (especially write operations)
- Support OIDC authentication when enabled
- Require admin privileges for destructive operations (remove, scale)
- Rate limit to prevent abuse
- Validate input parameters
- Return sanitized errors (no sensitive data in error messages)
---
## Error Handling
Standard error response format:
```json
{
"error": "ContainerNotFound",
"message": "Container 'nginx2' not found",
"details": {
"container": "nginx2",
"available_containers": ["nginx", "core-ai", "ollama"]
}
}
```
Common error codes:
- `400` - Bad Request (invalid parameters)
- `404` - Not Found (container/service doesn't exist)
- `409` - Conflict (invalid state transition)
- `500` - Internal Server Error (Portainer/Docker API failed)
- `503` - Service Unavailable (Portainer/Docker not accessible)
---
## Testing
Each endpoint should have:
- Unit tests (mock Portainer client)
- Integration tests (real Portainer/Docker)
- Error case tests (not found, permission denied, etc.)
- Performance tests (ensure response times < 2s)
---
## Questions / Decisions Needed
1. **Authentication**: Should container management require admin role, or allow read-only for all users?
2. **Rate Limiting**: What limits should be applied to prevent abuse?
3. **Caching**: Should container lists be cached? (TTL: 5s?)
4. **Async**: Should heavy operations (like logs) be async with job IDs?
5. **Webhooks**: Should operations emit events for monitoring?
---
## Notes
- All endpoints follow RESTful conventions
- Use existing PortainerClient methods where available
- Fall back to Docker socket when Portainer doesn't have data
- Log all operations with timestamps, user, and outcome
- Consider adding `/v1/infrastructure/containers/search` for fuzzy name matching
+68
View File
@@ -0,0 +1,68 @@
# Alembic Configuration for Core-API
#
# Database migrations using async SQLAlchemy
[alembic]
# Path to migration scripts
script_location = alembic
# Template for migration file names
file_template = %%(year)d%%(month).2d%%(day).2d_%%(hour).2d%%(minute).2d_%%(rev)s_%%(slug)s
# Prepend sys.path with the project root
prepend_sys_path = .
# Timezone for revision creation date
timezone = UTC
# Max length of revision identifiers
truncate_slug_length = 40
# Set to 'true' to run in offline mode
revision_environment = false
# Set to 'true' for sqlalchemy.url to be from the environment
# We use env.py to get the URL from config.py instead
sqlalchemy.url =
[post_write_hooks]
# Black formatting on generated migration files
# hooks = black
# black.type = console_scripts
# black.entrypoint = black
# black.options = -l 100 REVISION_SCRIPT_FILENAME
# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
qualname =
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S
+21
View File
@@ -0,0 +1,21 @@
Alembic Migrations for Core-API
This directory contains database migrations managed by Alembic.
Commands:
# Generate a new migration (after changing models)
alembic revision --autogenerate -m "description"
# Apply all pending migrations
alembic upgrade head
# Rollback last migration
alembic downgrade -1
# View migration history
alembic history
# View current revision
alembic current
See https://alembic.sqlalchemy.org for more documentation.
+98
View File
@@ -0,0 +1,98 @@
"""
Alembic Environment Configuration
Async migration environment for SQLAlchemy 2.0 with asyncpg.
"""
import asyncio
from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context
# Import our models and config
from src.config import get_settings
from src.db.database import Base
# Import all models to ensure they're registered with Base.metadata
from src.db.models import User, Role, UserRole, UserPreferences, ApiKey # noqa: F401
# Alembic Config object
config = context.config
# Interpret the config file for Python logging
if config.config_file_name is not None:
fileConfig(config.config_file_name)
# Model metadata for autogenerate support
target_metadata = Base.metadata
# Get database URL from our settings
settings = get_settings()
db_url = settings.database_url
if db_url.startswith("postgresql://"):
db_url = db_url.replace("postgresql://", "postgresql+asyncpg://", 1)
def run_migrations_offline() -> None:
"""
Run migrations in 'offline' mode.
Generates SQL script without connecting to the database.
"""
context.configure(
url=db_url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def do_run_migrations(connection: Connection) -> None:
"""
Run migrations with the given connection.
"""
context.configure(
connection=connection,
target_metadata=target_metadata,
)
with context.begin_transaction():
context.run_migrations()
async def run_async_migrations() -> None:
"""
Run migrations in 'online' mode with async engine.
"""
configuration = config.get_section(config.config_ini_section) or {}
configuration["sqlalchemy.url"] = db_url
connectable = async_engine_from_config(
configuration,
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
async with connectable.connect() as connection:
await connection.run_sync(do_run_migrations)
await connectable.dispose()
def run_migrations_online() -> None:
"""
Run migrations in 'online' mode.
"""
asyncio.run(run_async_migrations())
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()
+25
View File
@@ -0,0 +1,25 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
# revision identifiers, used by Alembic.
revision: str = ${repr(up_revision)}
down_revision: Union[str, None] = ${repr(down_revision)}
branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
def upgrade() -> None:
${upgrades if upgrades else "pass"}
def downgrade() -> None:
${downgrades if downgrades else "pass"}
@@ -0,0 +1,138 @@
"""Create auth tables
Revision ID: 001
Revises:
Create Date: 2026-01-01
Creates the initial authentication and authorization tables:
- users: User accounts synced from Authentik
- roles: Domain-scoped permission roles
- user_roles: User-Role association table
- user_preferences: User settings and preferences
- api_keys: API key authentication
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision: str = "001"
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# Users table
op.create_table(
"users",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("authentik_id", postgresql.UUID(as_uuid=True), nullable=False),
sa.Column("email", sa.String(255), nullable=False),
sa.Column("name", sa.String(255), nullable=False),
sa.Column("avatar_url", sa.String(500), nullable=True),
sa.Column("api_keys_enabled", sa.Boolean(), nullable=False, server_default="true"),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.Column("last_login", sa.DateTime(timezone=True), nullable=True),
)
op.create_index("ix_users_authentik_id", "users", ["authentik_id"], unique=True)
op.create_index("ix_users_email", "users", ["email"], unique=True)
# Roles table
op.create_table(
"roles",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("name", sa.String(100), nullable=False, comment="Role name in format domain:action"),
sa.Column("domain", sa.String(50), nullable=False, comment="Permission domain"),
sa.Column("action", sa.String(20), nullable=False, comment="Permission action"),
sa.Column("authentik_group", sa.String(255), nullable=True, comment="Corresponding Authentik group name"),
)
op.create_index("ix_roles_name", "roles", ["name"], unique=True)
op.create_index("ix_roles_domain", "roles", ["domain"])
op.create_index("ix_roles_authentik_group", "roles", ["authentik_group"], unique=True)
# User-Role association table
op.create_table(
"user_roles",
sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
sa.Column("role_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("roles.id", ondelete="CASCADE"), primary_key=True),
)
# User preferences table
op.create_table(
"user_preferences",
sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
sa.Column("theme", sa.String(20), nullable=False, server_default="system", comment="Theme preference"),
sa.Column("default_room", sa.String(50), nullable=False, server_default="front-hall", comment="Default room"),
sa.Column("preferences_json", postgresql.JSONB(), nullable=False, server_default="{}"),
)
# API keys table
op.create_table(
"api_keys",
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column("user_id", postgresql.UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
sa.Column("name", sa.String(100), nullable=False, comment="Human-readable key name"),
sa.Column("key_hash", sa.String(255), nullable=False, comment="SHA-256 hash of the API key"),
sa.Column("key_prefix", sa.String(8), nullable=False, comment="First 8 chars for identification"),
sa.Column("scopes", postgresql.ARRAY(sa.String()), nullable=True, comment="Optional scope restriction"),
sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("last_used_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
)
op.create_index("ix_api_keys_user_id", "api_keys", ["user_id"])
# Seed initial roles (domain:action combinations)
roles_table = sa.table(
"roles",
sa.column("id", postgresql.UUID),
sa.column("name", sa.String),
sa.column("domain", sa.String),
sa.column("action", sa.String),
sa.column("authentik_group", sa.String),
)
domains = [
"control-room",
"library",
"media",
"ai",
"housekeeper",
"developer",
"documents",
"gaming",
"admin",
]
actions = ["viewer", "user", "editor", "admin"]
roles_data = []
for domain in domains:
for action in actions:
role_name = f"{domain}:{action}"
authentik_group = f"tatlock-{domain}-{action}"
roles_data.append({
"id": sa.text("gen_random_uuid()"),
"name": role_name,
"domain": domain,
"action": action,
"authentik_group": authentik_group,
})
# Insert roles using raw SQL for UUID generation
for role in roles_data:
op.execute(
f"""
INSERT INTO roles (id, name, domain, action, authentik_group)
VALUES (gen_random_uuid(), '{role["name"]}', '{role["domain"]}', '{role["action"]}', '{role["authentik_group"]}')
"""
)
def downgrade() -> None:
op.drop_table("api_keys")
op.drop_table("user_preferences")
op.drop_table("user_roles")
op.drop_table("roles")
op.drop_table("users")
+1 -1
View File
@@ -1,6 +1,6 @@
[project] [project]
name = "core-api" name = "core-api"
version = "1.2.0" version = "1.4.5"
description = "Core Code API - Infrastructure management and tools API" description = "Core Code API - Infrastructure management and tools API"
readme = "README.md" readme = "README.md"
requires-python = ">=3.12" requires-python = ">=3.12"
+14 -8
View File
@@ -1,12 +1,13 @@
# FastAPI and ASGI server # FastAPI and ASGI server
fastapi~=0.115.0 fastapi>=0.115.0
uvicorn[standard]>=0.34.0 # Updated for google-adk compatibility starlette>=0.49.1 # CVE-2025-54121, CVE-2025-62727
pydantic>=2.11.1,<3.0.0 # Required for google-cloud-aiplatform[agent-engines] uvicorn[standard]>=0.34.0
pydantic>=2.11.1,<3.0.0
pydantic-settings>=2.10.1 pydantic-settings>=2.10.1
# HTTP client # HTTP client
httpx>=0.28.0 # Required for google-adk httpx>=0.28.0
python-socketio[asyncio_client]~=5.11.0 python-socketio[asyncio_client]>=5.14.0 # CVE-2025-61765
# Web scraping # Web scraping
beautifulsoup4~=4.12.0 beautifulsoup4~=4.12.0
@@ -22,6 +23,11 @@ pytz~=2024.1
dnspython~=2.7.0 dnspython~=2.7.0
# Authentication & Security # Authentication & Security
PyJWT[crypto]~=2.9.0 PyJWT[crypto]>=2.9.0
python-jose[cryptography]~=3.3.0 python-jose[cryptography]>=3.4.0 # CVE PYSEC-2024-232, PYSEC-2024-233
cryptography~=43.0.0 cryptography>=44.0.1 # CVE-2024-12797
# Database
sqlalchemy[asyncio]~=2.0.0
asyncpg>=0.30.0
alembic~=1.13.0
+22
View File
@@ -3,3 +3,25 @@ Authentication module for core-api
Provides OIDC/OAuth2 authentication via Authentik Provides OIDC/OAuth2 authentication via Authentik
""" """
from src.auth.oidc import (
get_current_user,
get_admin_user,
get_optional_user,
get_forward_auth_user,
get_forward_auth_admin,
oidc_config,
)
from src.auth.service import AuthService, get_auth_service
from src.auth.controller import auth_controller
__all__ = [
"get_current_user",
"get_admin_user",
"get_optional_user",
"get_forward_auth_user",
"get_forward_auth_admin",
"oidc_config",
"AuthService",
"get_auth_service",
"auth_controller",
]
+183
View File
@@ -0,0 +1,183 @@
"""
Authentication Controller
Provides authentication endpoints for OIDC token sync and user management.
"""
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import JSONResponse
from sqlalchemy.ext.asyncio import AsyncSession
from src.controllers.base import BaseController
from src.logging_config import get_logger
from src.db import get_async_session
from src.auth.schemas import AuthSyncRequest, AuthSyncResponse, UsersListResponse, BulkSyncResultSchema
from src.auth.service import AuthService
logger = get_logger(__name__)
class AuthController(BaseController):
"""
Controller for authentication operations
Provides endpoints for:
- Token synchronization (login)
- User profile retrieval
"""
def __init__(self):
super().__init__(prefix="/auth", tags=["Authentication"])
def create_router(self) -> APIRouter:
"""Create and configure the router"""
router = APIRouter(prefix=self.prefix, tags=self.tags)
@router.post(
"/sync",
summary="Sync user from OIDC token",
response_model=AuthSyncResponse,
responses={
200: {"description": "User synced successfully"},
401: {"description": "Invalid or expired token"},
503: {"description": "Authentication service unavailable"},
},
)
async def sync_user(
request: AuthSyncRequest,
session: AsyncSession = Depends(get_async_session),
) -> AuthSyncResponse:
"""
Synchronize user from OIDC access token
This endpoint should be called after the client obtains an access token
from Authentik. It:
1. Validates the token via Authentik's userinfo endpoint
2. Creates or updates the user in the database
3. Syncs roles from Authentik groups
4. Returns the user profile with roles and preferences
The client should store the returned user info for local use.
"""
service = AuthService(session)
try:
# Validate token with Authentik
token_info = await service.validate_token(request.access_token)
except ValueError as e:
logger.warning(f"Token validation failed: {e}")
raise HTTPException(status_code=401, detail=str(e))
# Sync user to database
user, is_new = await service.sync_user(token_info)
# Sync roles from groups
roles = await service.sync_roles(user, token_info.groups)
# Commit the transaction
await session.commit()
# Refresh to get relationships
await session.refresh(user, ["preferences"])
# Build response
return AuthSyncResponse(
user=service.user_to_schema(user),
roles=service.roles_to_schema(roles),
preferences=service.preferences_to_schema(user.preferences),
is_new_user=is_new,
)
@router.get(
"/users",
summary="List all users",
response_model=UsersListResponse,
responses={
200: {"description": "List of users"},
},
)
async def list_users(
search: Optional[str] = Query(None, description="Search by name or email"),
offset: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(50, ge=1, le=100, description="Maximum records to return"),
session: AsyncSession = Depends(get_async_session),
) -> UsersListResponse:
"""
List all users who have logged in via Authentik
Returns paginated list of users with their roles.
Supports search filtering by name or email.
"""
service = AuthService(session)
items, total = await service.list_users(
search=search,
offset=offset,
limit=limit,
)
return UsersListResponse(items=items, total=total)
@router.post(
"/users/sync-from-authentik",
summary="Bulk sync users from Authentik",
response_model=BulkSyncResultSchema,
responses={
200: {"description": "Sync completed"},
401: {"description": "Authentik API token invalid"},
503: {"description": "Authentik service unavailable"},
},
)
async def sync_users_from_authentik(
session: AsyncSession = Depends(get_async_session),
) -> BulkSyncResultSchema:
"""
Fetch all users from Authentik and sync to local database
This endpoint uses the Authentik admin API to fetch all users
and create/update them in the local database. Requires
AUTHENTIK_CORE_API_TOKEN to be configured.
Use this to initially populate users or to re-sync after
changes in Authentik.
"""
service = AuthService(session)
try:
result = await service.bulk_sync_from_authentik()
logger.info(
f"Bulk sync completed: {result.created} created, "
f"{result.updated} updated, {result.failed} failed"
)
return result
except ValueError as e:
logger.error(f"Bulk sync failed: {e}")
raise HTTPException(status_code=401, detail=str(e))
@router.get(
"/me",
summary="Get current user profile",
response_model=AuthSyncResponse,
responses={
200: {"description": "User profile"},
401: {"description": "Not authenticated"},
},
)
async def get_me(
session: AsyncSession = Depends(get_async_session),
) -> JSONResponse:
"""
Get the current authenticated user's profile
Note: This endpoint requires a valid session or API key.
For now, returns 501 Not Implemented until session management is added.
"""
# TODO: Implement with get_current_user dependency
raise HTTPException(
status_code=501,
detail="Not implemented - use /auth/sync with access token",
)
return router
# Create controller instance
auth_controller = AuthController()
+110
View File
@@ -0,0 +1,110 @@
"""
Authentication Schemas
Pydantic models for auth request/response payloads.
"""
import uuid
from datetime import datetime
from typing import Optional
from pydantic import Field
from src.base_schema import BaseSchema
class AuthSyncRequest(BaseSchema):
"""
Request payload for POST /auth/sync
The client sends this after obtaining an OIDC token from Authentik.
The access_token is validated against Authentik's userinfo endpoint.
"""
access_token: str = Field(
...,
description="OIDC access token from Authentik",
)
class RoleSchema(BaseSchema):
"""Role information in domain:action format"""
name: str = Field(..., description="Role name (e.g., 'control-room:admin')")
domain: str = Field(..., description="Permission domain (e.g., 'control-room')")
action: str = Field(..., description="Permission action (e.g., 'admin')")
class UserPreferencesSchema(BaseSchema):
"""User preferences"""
theme: str = Field(default="system", description="Theme preference: system, light, dark")
default_room: str = Field(default="front-hall", description="Default room for housekeeping")
preferences_json: dict = Field(default_factory=dict, description="Extended preferences")
class UserSchema(BaseSchema):
"""User information returned from sync"""
id: uuid.UUID = Field(..., description="Internal user ID")
authentik_id: uuid.UUID = Field(..., description="Authentik user ID")
email: str = Field(..., description="User email")
name: str = Field(..., description="Display name")
avatar_url: Optional[str] = Field(None, description="Profile picture URL")
created_at: datetime = Field(..., description="Account creation timestamp")
last_login: Optional[datetime] = Field(None, description="Last login timestamp")
class AuthSyncResponse(BaseSchema):
"""
Response from POST /auth/sync
Contains the synced user profile, roles, and preferences.
"""
user: UserSchema = Field(..., description="User profile")
roles: list[RoleSchema] = Field(..., description="User's permission roles")
preferences: UserPreferencesSchema = Field(..., description="User preferences")
is_new_user: bool = Field(..., description="True if user was just created")
class TokenInfoSchema(BaseSchema):
"""
Token information from Authentik userinfo endpoint
This is what Authentik returns when validating an access token.
"""
sub: str = Field(..., description="Subject (Authentik user ID)")
email: str = Field(..., description="User email")
name: Optional[str] = Field(None, description="Display name")
preferred_username: Optional[str] = Field(None, description="Username")
groups: list[str] = Field(default_factory=list, description="Group memberships")
picture: Optional[str] = Field(None, description="Profile picture URL")
class UserListItemSchema(BaseSchema):
"""User item for list display"""
id: uuid.UUID = Field(..., description="Internal user ID")
email: str = Field(..., description="User email")
name: str = Field(..., description="Display name")
avatar_url: Optional[str] = Field(None, description="Profile picture URL")
created_at: datetime = Field(..., description="Account creation timestamp")
last_login: Optional[datetime] = Field(None, description="Last login timestamp")
roles: list[str] = Field(default_factory=list, description="Role names")
class UsersListResponse(BaseSchema):
"""Response from GET /auth/users"""
items: list[UserListItemSchema] = Field(..., description="List of users")
total: int = Field(..., description="Total count of users")
class BulkSyncResultSchema(BaseSchema):
"""Result from bulk sync operation"""
created: int = Field(..., description="Number of users created")
updated: int = Field(..., description="Number of users updated")
failed: int = Field(..., description="Number of users that failed to sync")
total_in_authentik: int = Field(..., description="Total users in Authentik")
errors: list[str] = Field(default_factory=list, description="Error messages for failed syncs")
+477
View File
@@ -0,0 +1,477 @@
"""
Authentication Service
Business logic for user synchronization from Authentik.
"""
import uuid
from datetime import datetime, timezone
from typing import Optional
import httpx
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from src.config import get_settings
from src.logging_config import get_logger
from src.db.models import User, Role, UserPreferences
from src.auth.schemas import TokenInfoSchema, UserSchema, RoleSchema, UserPreferencesSchema, UserListItemSchema, BulkSyncResultSchema
logger = get_logger(__name__)
settings = get_settings()
class AuthService:
"""
Service for authentication and user synchronization
Handles:
- Token validation via Authentik userinfo endpoint
- User creation/update from OIDC claims
- Role synchronization from Authentik groups
"""
def __init__(self, session: AsyncSession):
"""
Initialize auth service
Args:
session: Async database session
"""
self.session = session
self.userinfo_url = f"{settings.authentik_url}/application/o/userinfo/"
async def validate_token(self, access_token: str) -> TokenInfoSchema:
"""
Validate access token via Authentik userinfo endpoint
Args:
access_token: OIDC access token
Returns:
Token info containing user claims
Raises:
ValueError: If token is invalid or expired
"""
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
self.userinfo_url,
headers={"Authorization": f"Bearer {access_token}"},
)
if response.status_code == 401:
raise ValueError("Invalid or expired token")
response.raise_for_status()
data = response.json()
logger.debug(f"Userinfo response: {data}")
return TokenInfoSchema(
sub=data.get("sub"),
email=data.get("email"),
name=data.get("name") or data.get("preferred_username"),
preferred_username=data.get("preferred_username"),
groups=data.get("groups", []),
picture=data.get("picture"),
)
except httpx.HTTPStatusError as e:
logger.error(f"Authentik userinfo request failed: {e}")
raise ValueError(f"Token validation failed: {e.response.status_code}")
except httpx.RequestError as e:
logger.error(f"Authentik userinfo request error: {e}")
raise ValueError("Authentication service unavailable")
async def sync_user(self, token_info: TokenInfoSchema) -> tuple[User, bool]:
"""
Create or update user from OIDC token info
Args:
token_info: Validated token information
Returns:
Tuple of (User, is_new_user)
"""
authentik_id = uuid.UUID(token_info.sub)
# Try to find existing user
stmt = (
select(User)
.options(selectinload(User.roles), selectinload(User.preferences))
.where(User.authentik_id == authentik_id)
)
result = await self.session.execute(stmt)
user = result.scalar_one_or_none()
is_new = user is None
if is_new:
# Create new user
user = User(
authentik_id=authentik_id,
email=token_info.email,
name=token_info.name or token_info.email,
avatar_url=token_info.picture,
last_login=datetime.now(timezone.utc),
)
self.session.add(user)
await self.session.flush() # Get the user ID
# Create default preferences
preferences = UserPreferences(user_id=user.id)
self.session.add(preferences)
logger.info(f"Created new user: {token_info.email}")
else:
# Update existing user
user.email = token_info.email
user.name = token_info.name or token_info.email
user.avatar_url = token_info.picture
user.last_login = datetime.now(timezone.utc)
logger.info(f"Updated existing user: {token_info.email}")
await self.session.flush()
return user, is_new
async def sync_roles(self, user: User, groups: list[str]) -> list[Role]:
"""
Synchronize user roles from Authentik groups
Maps Authentik groups (e.g., 'tatlock-control-room-admin')
to application roles (e.g., 'control-room:admin').
Args:
user: User to sync roles for
groups: List of Authentik group names
Returns:
List of synced Role objects
"""
# Get all roles that match the user's Authentik groups
stmt = select(Role).where(Role.authentik_group.in_(groups))
result = await self.session.execute(stmt)
matching_roles = list(result.scalars().all())
# Clear existing roles and set new ones
user.roles = matching_roles
role_names = [r.name for r in matching_roles]
logger.info(f"Synced roles for {user.email}: {role_names}")
return matching_roles
def user_to_schema(self, user: User) -> UserSchema:
"""Convert User model to schema"""
return UserSchema(
id=user.id,
authentik_id=user.authentik_id,
email=user.email,
name=user.name,
avatar_url=user.avatar_url,
created_at=user.created_at,
last_login=user.last_login,
)
def roles_to_schema(self, roles: list[Role]) -> list[RoleSchema]:
"""Convert Role models to schemas"""
return [
RoleSchema(name=r.name, domain=r.domain, action=r.action)
for r in roles
]
def preferences_to_schema(self, preferences: Optional[UserPreferences]) -> UserPreferencesSchema:
"""Convert UserPreferences model to schema"""
if preferences is None:
return UserPreferencesSchema()
return UserPreferencesSchema(
theme=preferences.theme,
default_room=preferences.default_room,
preferences_json=preferences.preferences_json or {},
)
async def list_users(
self,
search: Optional[str] = None,
offset: int = 0,
limit: int = 50,
) -> tuple[list[UserListItemSchema], int]:
"""
List all users with optional search and pagination
Args:
search: Optional search query (matches name or email)
offset: Number of records to skip
limit: Maximum number of records to return
Returns:
Tuple of (list of user schemas, total count)
"""
from sqlalchemy import func
# Base query with roles loaded
base_query = select(User).options(selectinload(User.roles))
# Apply search filter if provided
if search:
search_filter = f"%{search}%"
base_query = base_query.where(
(User.name.ilike(search_filter)) | (User.email.ilike(search_filter))
)
# Get total count
count_query = select(func.count()).select_from(base_query.subquery())
total_result = await self.session.execute(count_query)
total = total_result.scalar() or 0
# Apply pagination and ordering
query = base_query.order_by(User.name).offset(offset).limit(limit)
result = await self.session.execute(query)
users = list(result.scalars().all())
# Convert to schemas
items = [
UserListItemSchema(
id=user.id,
email=user.email,
name=user.name,
avatar_url=user.avatar_url,
created_at=user.created_at,
last_login=user.last_login,
roles=[role.name for role in user.roles],
)
for user in users
]
return items, total
def _extract_cookie(self, headers: httpx.Headers, cookie_name: str) -> str:
"""Extract a specific cookie value from Set-Cookie headers"""
import re
for header in headers.get_list('set-cookie'):
if header.startswith(f'{cookie_name}='):
match = re.match(rf'{cookie_name}=([^;]+)', header)
if match:
return match.group(1)
return ""
async def _authentik_session_login(self, client: httpx.AsyncClient) -> str:
"""
Authenticate with Authentik using the flow API to establish a session
Authentik's flow API requires:
1. Cookie persistence between requests (manually handled due to domain restrictions)
2. X-authentik-CSRF header set to the authentik_csrf cookie value
3. Multi-stage flow handling (identification -> password -> done)
Args:
client: httpx client
Returns:
Session cookie value for subsequent API calls
Raises:
ValueError: If authentication fails
"""
flow_url = f"{settings.authentik_url}/api/v3/flows/executor/default-authentication-flow/"
# Step 1: Get the initial flow challenge (this sets the session and csrf cookies)
resp = await client.get(flow_url, headers={"Accept": "application/json"})
resp.raise_for_status()
data = resp.json()
# Extract cookies manually from Set-Cookie headers (bypasses domain restrictions)
session_cookie = self._extract_cookie(resp.headers, "authentik_session")
csrf_cookie = self._extract_cookie(resp.headers, "authentik_csrf")
logger.debug(f"Flow initial: component={data.get('component')}, session={bool(session_cookie)}, csrf={bool(csrf_cookie)}")
# Build headers with manual cookie and CSRF token
def build_headers():
hdrs = {
"Accept": "application/json",
"Content-Type": "application/json",
"Cookie": f"authentik_session={session_cookie}",
}
if csrf_cookie:
hdrs["Cookie"] += f"; authentik_csrf={csrf_cookie}"
hdrs["X-authentik-CSRF"] = csrf_cookie
return hdrs
# Step 2: Handle identification stage - submit username
if data.get("component") == "ak-stage-identification":
resp = await client.post(
flow_url,
json={"uid_field": settings.authentik_username},
headers=build_headers(),
)
resp.raise_for_status()
data = resp.json()
# Update session cookie if new one received
new_session = self._extract_cookie(resp.headers, "authentik_session")
if new_session:
session_cookie = new_session
logger.debug(f"After username: component={data.get('component')}")
# Step 3: Handle password stage if required
if data.get("component") == "ak-stage-password":
resp = await client.post(
flow_url,
json={"password": settings.authentik_password},
headers=build_headers(),
)
resp.raise_for_status()
data = resp.json()
# Update session cookie if new one received
new_session = self._extract_cookie(resp.headers, "authentik_session")
if new_session:
session_cookie = new_session
logger.debug(f"After password: component={data.get('component')}")
# Check for access denied
if data.get("component") == "ak-stage-access-denied":
raise ValueError("Authentik authentication failed: access denied")
# Check for redirect (successful auth)
if data.get("component") == "xak-flow-redirect" or data.get("to"):
logger.info("Successfully authenticated with Authentik via flow")
return session_cookie
# If we're still in identification stage, the username might be wrong
if data.get("component") == "ak-stage-identification":
response_errors = data.get("response_errors", {})
raise ValueError(f"Authentication stuck at identification stage: {response_errors}")
logger.info(f"Authentik flow completed with component: {data.get('component')}")
return session_cookie
async def bulk_sync_from_authentik(self) -> BulkSyncResultSchema:
"""
Fetch all users from Authentik admin API and sync to local database
Returns:
BulkSyncResultSchema with counts of created/updated/failed users
"""
if not settings.authentik_username or not settings.authentik_password:
raise ValueError("AUTHENTIK_USERNAME and AUTHENTIK_PASSWORD must be configured")
created = 0
updated = 0
failed = 0
errors = []
total_in_authentik = 0
# Step 1: Fetch all user data from Authentik API
authentik_users = []
try:
async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client:
# Authenticate with Authentik to get session cookie
session_cookie = await self._authentik_session_login(client)
# Fetch users from Authentik admin API using session cookie
response = await client.get(
f"{settings.authentik_url}/api/v3/core/users/",
params={"page_size": 500},
headers={
"Accept": "application/json",
"Cookie": f"authentik_session={session_cookie}",
},
)
if response.status_code == 401:
raise ValueError("Authentik API token is invalid or expired")
response.raise_for_status()
data = response.json()
authentik_users = data.get("results", [])
total_in_authentik = data.get("pagination", {}).get("count", len(authentik_users))
except httpx.HTTPStatusError as e:
raise ValueError(f"Authentik API error: {e.response.status_code}")
except httpx.RequestError as e:
raise ValueError(f"Failed to connect to Authentik: {str(e)}")
# Step 2: Sync users to database (outside of httpx context to avoid greenlet issues)
for auth_user in authentik_users:
try:
# Skip service accounts and inactive users
if auth_user.get("type") in ("service_account", "internal_service_account"):
continue
if not auth_user.get("is_active", True):
continue
# Extract user data from Authentik
authentik_id = uuid.UUID(auth_user["uuid"])
email = auth_user.get("email") or f"{auth_user['username']}@local"
name = auth_user.get("name") or auth_user.get("username", "Unknown")
avatar_url = auth_user.get("avatar")
# Get user's groups for role mapping
groups = []
groups_summary = auth_user.get("groups_obj", [])
for group in groups_summary:
groups.append(group.get("name", ""))
# Check if user exists
stmt = select(User).where(User.authentik_id == authentik_id)
result = await self.session.execute(stmt)
user = result.scalar_one_or_none()
if user is None:
# Create new user
user = User(
authentik_id=authentik_id,
email=email,
name=name,
avatar_url=avatar_url,
)
self.session.add(user)
await self.session.flush()
# Create default preferences
preferences = UserPreferences(user_id=user.id)
self.session.add(preferences)
created += 1
logger.info(f"Created user from Authentik: {email}")
else:
# Update existing user
user.email = email
user.name = name
user.avatar_url = avatar_url
updated += 1
logger.info(f"Updated user from Authentik: {email}")
# Sync roles from groups
await self.sync_roles(user, groups)
except Exception as e:
failed += 1
error_msg = f"Failed to sync user {auth_user.get('username', 'unknown')}: {str(e)}"
errors.append(error_msg)
logger.warning(error_msg)
# Commit all changes
await self.session.commit()
return BulkSyncResultSchema(
created=created,
updated=updated,
failed=failed,
total_in_authentik=total_in_authentik,
errors=errors,
)
# Factory function for dependency injection
def get_auth_service(session: AsyncSession) -> AuthService:
"""Create AuthService instance with database session"""
return AuthService(session)
-197
View File
@@ -1,197 +0,0 @@
"""
Core-AI HTTP Client
Provides interface to Core-AI service for AI performance metrics.
"""
import httpx
from typing import Optional, Dict, List, Any
from src.logging_config import get_logger
from src.config import get_settings
logger = get_logger(__name__)
settings = get_settings()
class CoreAIClient:
"""
HTTP client for Core-AI service
Provides access to AI performance metrics, tool execution stats,
and memory system monitoring.
"""
def __init__(
self,
base_url: Optional[str] = None,
timeout: int = 10
):
"""
Initialize Core-AI client
Args:
base_url: Core-AI base URL (default from settings)
timeout: Request timeout in seconds
"""
self.base_url = (base_url or getattr(settings, 'core_ai_base_url', 'http://core-ai:8086')).rstrip("/")
self.timeout = timeout
self.client = httpx.AsyncClient(timeout=self.timeout)
async def close(self):
"""Close the HTTP client"""
await self.client.aclose()
async def health_check(self) -> bool:
"""
Check if Core-AI service is accessible
Returns:
True if accessible, False otherwise
"""
try:
response = await self.client.get(f"{self.base_url}/health")
return response.status_code == 200
except Exception as e:
logger.error(f"Core-AI health check failed: {e}")
return False
async def get_metrics(self) -> Dict[str, Any]:
"""
Get comprehensive AI performance metrics
Returns:
Dict with agent performance, tool execution, memory stats
Example:
{
"uptime_seconds": 3600,
"timestamp": "2025-12-03T20:00:00Z",
"agent": {
"total_requests": 100,
"avg_response_time_ms": 1250.5,
"p95_response_time_ms": 3200.0,
...
},
"tools": {
"total_calls": 250,
"success_rate": 0.98,
"top_tools": {...}
},
"memory": {
"tier1_hit_rate": 0.85,
...
},
...
}
"""
try:
response = await self.client.get(f"{self.base_url}/metrics")
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
logger.error(f"Failed to get metrics: HTTP {e.response.status_code}")
raise
except Exception as e:
logger.error(f"Failed to get metrics: {e}")
raise
async def get_recent_errors(self, limit: int = 20) -> List[Dict[str, Any]]:
"""
Get recent request errors
Args:
limit: Maximum number of errors to return
Returns:
List of error records with timestamps
Example:
[
{
"timestamp": "2025-12-03T19:45:12Z",
"agent_type": "pydantic",
"error": "Connection timeout",
"duration_ms": 5000
},
...
]
"""
try:
response = await self.client.get(
f"{self.base_url}/metrics/errors",
params={"limit": limit}
)
response.raise_for_status()
data = response.json()
return data.get("errors", [])
except Exception as e:
logger.error(f"Failed to get recent errors: {e}")
raise
async def get_tool_failures(self, limit: int = 20) -> List[Dict[str, Any]]:
"""
Get recent tool execution failures
Args:
limit: Maximum number of failures to return
Returns:
List of tool failure records
Example:
[
{
"timestamp": "2025-12-03T19:50:30Z",
"tool_name": "list_containers",
"error": "Connection refused",
"duration_ms": 150
},
...
]
"""
try:
response = await self.client.get(
f"{self.base_url}/metrics/tool-failures",
params={"limit": limit}
)
response.raise_for_status()
data = response.json()
return data.get("failures", [])
except Exception as e:
logger.error(f"Failed to get tool failures: {e}")
raise
async def reset_metrics(self) -> bool:
"""
Reset all metrics (admin operation)
Returns:
True if successful
"""
try:
response = await self.client.post(f"{self.base_url}/metrics/reset")
response.raise_for_status()
logger.info("Successfully reset Core-AI metrics")
return True
except Exception as e:
logger.error(f"Failed to reset metrics: {e}")
raise
async def __aenter__(self):
"""Async context manager entry"""
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Async context manager exit"""
await self.close()
# Singleton instance
_ai_client: Optional[CoreAIClient] = None
def get_ai_client() -> CoreAIClient:
"""Get singleton Core-AI client instance"""
global _ai_client
if _ai_client is None:
_ai_client = CoreAIClient()
return _ai_client
+158 -102
View File
@@ -210,6 +210,152 @@ class PortainerClient:
response.raise_for_status() response.raise_for_status()
return True return True
async def get_stack_file(self, stack_id: int) -> str:
"""
Get the compose file content for a stack
Args:
stack_id: Stack identifier
Returns:
Docker Compose YAML content as string
"""
async with httpx.AsyncClient(timeout=self.timeout) as client:
response = await client.get(
f"{self.base_url}/api/stacks/{stack_id}/file",
headers=self._get_headers()
)
response.raise_for_status()
data = response.json()
return data.get("StackFileContent", "")
async def redeploy_stack(
self,
stack_id: int,
endpoint_id: int,
pull_image: bool = False
) -> Dict[str, Any]:
"""
Redeploy a stack with its current configuration
Args:
stack_id: Stack identifier
endpoint_id: Portainer endpoint
pull_image: Pull latest images before deployment
Returns:
Updated stack details
"""
# Get current stack file content
stack_content = await self.get_stack_file(stack_id)
# Get current stack to preserve env vars
stack = await self.get_stack(stack_id)
env_vars = stack.get("Env", [])
payload = {
"stackFileContent": stack_content,
"env": env_vars,
"prune": False,
"pullImage": pull_image
}
async with httpx.AsyncClient(timeout=self.timeout) as client:
response = await client.put(
f"{self.base_url}/api/stacks/{stack_id}",
headers=self._get_headers(),
params={"endpointId": endpoint_id},
json=payload
)
response.raise_for_status()
return response.json()
async def update_stack_env(
self,
stack_id: int,
endpoint_id: int,
env_vars: List[Dict[str, str]]
) -> Dict[str, Any]:
"""
Update stack environment variables
Args:
stack_id: Stack identifier
endpoint_id: Portainer endpoint
env_vars: List of {"name": "VAR_NAME", "value": "var_value"} dicts
Returns:
Updated stack details
"""
# Get current stack file content (required for update)
stack_content = await self.get_stack_file(stack_id)
payload = {
"stackFileContent": stack_content,
"env": env_vars,
"prune": False,
"pullImage": False
}
async with httpx.AsyncClient(timeout=self.timeout) as client:
response = await client.put(
f"{self.base_url}/api/stacks/{stack_id}",
headers=self._get_headers(),
params={"endpointId": endpoint_id},
json=payload
)
response.raise_for_status()
return response.json()
async def delete_container(
self,
endpoint_id: int,
container_id: str,
force: bool = False
) -> bool:
"""
Delete a container
Args:
endpoint_id: Portainer endpoint identifier
container_id: Container ID or name
force: Force remove running container
Returns:
True if successful
"""
params = {"force": "true" if force else "false"}
async with httpx.AsyncClient(timeout=self.timeout) as client:
response = await client.delete(
f"{self.base_url}/api/endpoints/{endpoint_id}/docker/containers/{container_id}",
headers=self._get_headers(),
params=params
)
response.raise_for_status()
logger.info(f"Deleted container {container_id}")
return True
async def restart_container(self, endpoint_id: int, container_id: str) -> bool:
"""
Restart a container
Args:
endpoint_id: Portainer endpoint identifier
container_id: Container ID or name
Returns:
True if successful
"""
async with httpx.AsyncClient(timeout=self.timeout) as client:
response = await client.post(
f"{self.base_url}/api/endpoints/{endpoint_id}/docker/containers/{container_id}/restart",
headers=self._get_headers()
)
response.raise_for_status()
logger.info(f"Restarted container {container_id}")
return True
async def get_containers(self, endpoint_id: int, all_containers: bool = True) -> List[Dict[str, Any]]: async def get_containers(self, endpoint_id: int, all_containers: bool = True) -> List[Dict[str, Any]]:
""" """
List containers on a specific endpoint List containers on a specific endpoint
@@ -292,70 +438,14 @@ class PortainerClient:
return True return True
# ======================================================================== # ========================================================================
# Docker Socket Fallback (for containers not managed by Portainer) # Helper methods for agent tools (auto-detect endpoint)
# ========================================================================
async def _list_containers_via_socket(self, all_containers: bool = True) -> List[Dict[str, Any]]:
"""
Fallback: List containers directly via Docker socket
Used when Portainer API doesn't return complete data (e.g., containers
started outside Portainer, AMP game servers, etc.)
Args:
all_containers: Include stopped containers
Returns:
List of container details in Docker API format
"""
try:
# Docker socket is mounted at /var/run/docker.sock
# Use httpx with unix socket transport
transport = httpx.AsyncHTTPTransport(uds="/var/run/docker.sock")
async with httpx.AsyncClient(transport=transport, timeout=10) as client:
params = {"all": 1 if all_containers else 0}
response = await client.get(
"http://localhost/v1.41/containers/json",
params=params
)
response.raise_for_status()
return response.json()
except Exception as e:
logger.warning(f"Docker socket fallback failed: {e}")
return []
async def _inspect_container_via_socket(self, container_id_or_name: str) -> Optional[Dict[str, Any]]:
"""
Fallback: Inspect container directly via Docker socket
Args:
container_id_or_name: Container ID or name
Returns:
Container details or None
"""
try:
transport = httpx.AsyncHTTPTransport(uds="/var/run/docker.sock")
async with httpx.AsyncClient(transport=transport, timeout=10) as client:
response = await client.get(
f"http://localhost/v1.41/containers/{container_id_or_name}/json"
)
response.raise_for_status()
return response.json()
except Exception as e:
logger.warning(f"Docker socket inspect fallback failed for '{container_id_or_name}': {e}")
return None
# ========================================================================
# Helper methods for agent tools (auto-detect endpoint + fallback)
# ======================================================================== # ========================================================================
async def list_containers(self, all_containers: bool = True) -> List[Dict[str, Any]]: async def list_containers(self, all_containers: bool = True) -> List[Dict[str, Any]]:
""" """
List containers using auto-detected endpoint with Docker socket fallback List containers using auto-detected endpoint
This is a convenience wrapper that automatically uses the first/default endpoint. This is a convenience wrapper that automatically uses the first/default endpoint.
If Portainer doesn't have complete data, falls back to Docker socket.
Args: Args:
all_containers: Include stopped containers (default: True) all_containers: Include stopped containers (default: True)
@@ -363,34 +453,18 @@ class PortainerClient:
Returns: Returns:
List of container details List of container details
""" """
try:
# Try Portainer first
endpoints = await self.get_endpoints() endpoints = await self.get_endpoints()
if endpoints: if not endpoints:
raise RuntimeError("No Portainer endpoints available")
endpoint_id = endpoints[0]["Id"] endpoint_id = endpoints[0]["Id"]
containers = await self.get_containers(endpoint_id, all_containers) return await self.get_containers(endpoint_id, all_containers)
if containers:
return containers
# Fallback to Docker socket
logger.info("Portainer returned no containers, trying Docker socket fallback...")
return await self._list_containers_via_socket(all_containers)
except Exception as e:
logger.error(f"Error listing containers: {e}")
# Try fallback even on exception
try:
return await self._list_containers_via_socket(all_containers)
except Exception as fallback_error:
logger.error(f"Fallback also failed: {fallback_error}")
return []
async def inspect_container(self, container_name: str) -> Optional[Dict[str, Any]]: async def inspect_container(self, container_name: str) -> Optional[Dict[str, Any]]:
""" """
Inspect a container by name using auto-detected endpoint with Docker socket fallback Inspect a container by name using auto-detected endpoint
This is a convenience wrapper that automatically uses the first/default endpoint. This is a convenience wrapper that automatically uses the first/default endpoint.
If Portainer doesn't find the container, falls back to Docker socket.
Args: Args:
container_name: Container name (e.g., "jellyfin", "ollama") container_name: Container name (e.g., "jellyfin", "ollama")
@@ -398,43 +472,25 @@ class PortainerClient:
Returns: Returns:
Container details or None if not found Container details or None if not found
""" """
try:
# Try Portainer first
endpoints = await self.get_endpoints() endpoints = await self.get_endpoints()
if endpoints: if not endpoints:
raise RuntimeError("No Portainer endpoints available")
endpoint_id = endpoints[0]["Id"] endpoint_id = endpoints[0]["Id"]
# First list all containers to find the one matching the name # List all containers to find the one matching the name
all_containers = await self.get_containers(endpoint_id, all_containers=True) all_containers = await self.get_containers(endpoint_id, all_containers=True)
matching_container = None
for container in all_containers: for container in all_containers:
# Container names come as array like ['/jellyfin'] # Container names come as array like ['/jellyfin']
names = container.get('Names', []) names = container.get('Names', [])
for name in names: for name in names:
clean_name = name.lstrip('/') clean_name = name.lstrip('/')
if clean_name == container_name or clean_name.lower() == container_name.lower(): if clean_name == container_name or clean_name.lower() == container_name.lower():
matching_container = container
break
if matching_container:
break
if matching_container:
# Get detailed info using container ID # Get detailed info using container ID
container_id = matching_container['Id'] container_id = container['Id']
return await self.get_container(endpoint_id, container_id) return await self.get_container(endpoint_id, container_id)
# Not found in Portainer, try Docker socket fallback
logger.info(f"Container '{container_name}' not found in Portainer, trying Docker socket fallback...")
return await self._inspect_container_via_socket(container_name)
except Exception as e:
logger.error(f"Error inspecting container '{container_name}': {e}")
# Try fallback even on exception
try:
return await self._inspect_container_via_socket(container_name)
except Exception as fallback_error:
logger.error(f"Fallback also failed: {fallback_error}")
return None return None
+35 -46
View File
@@ -1,5 +1,8 @@
""" """
Global configuration for Core Code API Global configuration for Core Code API
All configuration is loaded from environment variables or .env file.
See .env.example for available settings.
""" """
import tomllib import tomllib
from pathlib import Path from pathlib import Path
@@ -20,29 +23,6 @@ def _get_version_from_pyproject() -> str:
__version__ = _get_version_from_pyproject() __version__ = _get_version_from_pyproject()
# Import infrastructure credentials from gitignored module
try:
from src.credentials import (
PORTAINER_URL, PORTAINER_API_KEY,
NPM_URL, NPM_EMAIL, NPM_PASSWORD,
BRAVE_SEARCH_API_KEY,
GOOGLE_SEARCH_API_KEY, GOOGLE_SEARCH_ENGINE_ID,
HOMEASSISTANT_URL, HOMEASSISTANT_TOKEN
)
except ImportError:
# Fallback to empty strings if credentials.py doesn't exist
# (e.g., fresh clone before credentials setup)
PORTAINER_URL = "http://localhost:8001"
PORTAINER_API_KEY = ""
NPM_URL = "http://localhost:81"
NPM_EMAIL = ""
NPM_PASSWORD = ""
BRAVE_SEARCH_API_KEY = ""
GOOGLE_SEARCH_API_KEY = ""
GOOGLE_SEARCH_ENGINE_ID = ""
HOMEASSISTANT_URL = "http://localhost:8123"
HOMEASSISTANT_TOKEN = ""
class Settings(BaseSettings): class Settings(BaseSettings):
"""Global application settings""" """Global application settings"""
@@ -66,15 +46,13 @@ class Settings(BaseSettings):
log_level: str = "DEBUG" log_level: str = "DEBUG"
# Ollama Configuration (for AI orchestration) # Ollama Configuration (for AI orchestration)
ollama_base_url: str = "http://ollama:11434" ollama_base_url: str # Required - set OLLAMA_BASE_URL in .env
ollama_timeout: int = 300 # 5 minutes ollama_timeout: int = 300 # 5 minutes
# Model Configuration # Model Configuration
default_model: str = "mistral-tools:7b" default_model: str = "mistral-nemo-large:latest"
agent_model: str = "gemma2:9b-instruct-q5_K_M" # Must support tool calling with ADK (~4GB VRAM) agent_model: str = "mistral-nemo-large:latest" # Must support tool calling with ADK (~4GB VRAM)
lightweight_models: str = "gemma3-tools:1b,phi3:mini" code_models: str = "mistral-nemo-large:latest"
heavy_models: str = "mistral:7b,gemma2:9b,gemma3:12b,mixtral:8x7b"
code_models: str = "codestral:latest,codegemma:latest"
# Previous config (gemma3:12b used ~10GB VRAM) # Previous config (gemma3:12b used ~10GB VRAM)
# default_model: str = "gemma3:12b" # default_model: str = "gemma3:12b"
# agent_model: str = "gemma3:12b" # agent_model: str = "gemma3:12b"
@@ -109,35 +87,45 @@ class Settings(BaseSettings):
embedding_batch_size: int = 32 embedding_batch_size: int = 32
# Search Configuration # Search Configuration
search_provider: str = "google" # Options: google, brave, searxng, duckduckgo search_provider: str = "searxng"
searxng_url: str = "http://searxng:8080" # For future self-hosted SearxNG searxng_url: str # Required - set SEARXNG_URL in .env
# Search API Keys (from credentials.py) # Infrastructure Management (Portainer)
brave_search_api_key: str = BRAVE_SEARCH_API_KEY # https://brave.com/search/api/ portainer_url: str # Required - set PORTAINER_URL in .env
google_search_api_key: str = GOOGLE_SEARCH_API_KEY # https://console.cloud.google.com/ portainer_api_key: str # Required - set PORTAINER_API_KEY in .env
google_search_engine_id: str = GOOGLE_SEARCH_ENGINE_ID # Custom Search Engine ID
# Infrastructure Management (from credentials.py) # Infrastructure Management (Nginx Proxy Manager)
portainer_url: str = PORTAINER_URL npm_url: str # Required - set NPM_URL in .env
portainer_api_key: str = PORTAINER_API_KEY npm_email: str # Required - set NPM_EMAIL in .env
npm_password: str # Required - set NPM_PASSWORD in .env
npm_url: str = NPM_URL
npm_email: str = NPM_EMAIL
npm_password: str = NPM_PASSWORD
# Home Assistant Configuration # Home Assistant Configuration
homeassistant_url: str = HOMEASSISTANT_URL homeassistant_url: str # Required - set HOMEASSISTANT_URL in .env
homeassistant_token: str = HOMEASSISTANT_TOKEN homeassistant_token: str # Required - set HOMEASSISTANT_TOKEN in .env
homeassistant_timeout: int = 30 homeassistant_timeout: int = 30
# Core-AI Service (AI performance metrics) # PostgreSQL Database
core_ai_base_url: str = "http://core-ai:8086" postgres_host: str # Required - set POSTGRES_HOST in .env (e.g., localhost:5432)
postgres_user: str = "core_api"
postgres_password: str # Required - set POSTGRES_PASSWORD in .env
postgres_database: str = "core_api"
@property
def database_url(self) -> str:
"""Construct database URL from components"""
return f"postgresql://{self.postgres_user}:{self.postgres_password}@{self.postgres_host}/{self.postgres_database}"
# OIDC Authentication (Authentik) # OIDC Authentication (Authentik)
oidc_enabled: bool = False # Set to True to require authentication oidc_enabled: bool = False # Set to True to require authentication
oidc_issuer: str = "https://auth.schweitz.net/application/o/core-api/" oidc_issuer: str = "https://auth.schweitz.net/application/o/core-api/"
oidc_audience: str = "core-api" oidc_audience: str = "core-api"
# Authentik API (for token validation and user management)
# Must use domain name (not IP) when AUTHENTIK_COOKIE_DOMAIN is set
authentik_url: str = "https://auth.schweitz.net" # Authentik base URL
authentik_username: str = "" # Admin username for API access (AUTHENTIK_USERNAME env var)
authentik_password: str = "" # Admin password for API access (AUTHENTIK_PASSWORD env var)
@property @property
def model_aliases(self) -> dict: def model_aliases(self) -> dict:
"""Computed property for model aliases""" """Computed property for model aliases"""
@@ -163,6 +151,7 @@ class Settings(BaseSettings):
class Config: class Config:
env_file = ".env" env_file = ".env"
case_sensitive = False case_sensitive = False
extra = "ignore" # Ignore extra env vars not defined in Settings
@lru_cache() @lru_cache()
-219
View File
@@ -1,219 +0,0 @@
"""
AI Metrics Proxy Controller
Provides proxy endpoints to Core-AI service metrics.
Allows external access to AI performance stats via core-api.
"""
from fastapi import APIRouter, HTTPException
from typing import Dict, List, Any
from src.clients.ai_client import get_ai_client
from src.logging_config import get_logger
logger = get_logger(__name__)
# Create router
router = APIRouter(
prefix="/ai",
tags=["AI Metrics"]
)
@router.get(
"/health",
summary="Check Core-AI service health",
description="Verify that the Core-AI service is accessible and responding"
)
async def ai_health_check():
"""
Check if Core-AI service is healthy
Returns:
Health status and availability
"""
try:
ai_client = get_ai_client()
is_healthy = await ai_client.health_check()
return {
"service": "core-ai",
"status": "healthy" if is_healthy else "unhealthy",
"accessible": is_healthy
}
except Exception as e:
logger.error(f"AI health check failed: {e}")
return {
"service": "core-ai",
"status": "error",
"accessible": False,
"error": str(e)
}
@router.get(
"/metrics",
response_model=Dict[str, Any],
summary="Get comprehensive AI performance metrics",
description="Returns detailed metrics including agent performance, tool execution stats, memory system metrics, and user activity"
)
async def get_ai_metrics():
"""
Proxy endpoint for Core-AI metrics
Returns comprehensive AI performance data:
- Agent request statistics (total, by type, response times)
- Response time percentiles (p50, p95, p99)
- Tool execution metrics (calls, success rates, durations)
- Memory system statistics (cache hits, consolidations)
- User activity tracking
- Concurrency metrics
Returns:
Dict with all collected metrics
Raises:
HTTPException: If Core-AI is unreachable or returns error
"""
try:
ai_client = get_ai_client()
metrics = await ai_client.get_metrics()
return metrics
except Exception as e:
logger.error(f"Failed to fetch AI metrics: {e}")
raise HTTPException(
status_code=503,
detail=f"Core-AI service unavailable: {str(e)}"
)
@router.get(
"/metrics/errors",
response_model=Dict[str, Any],
summary="Get recent request errors",
description="Returns recent AI agent request errors with timestamps and details"
)
async def get_ai_errors(limit: int = 20):
"""
Get recent AI request errors
Args:
limit: Maximum number of errors to return (default: 20)
Returns:
Dict with error list and total count
Example response:
{
"errors": [
{
"timestamp": "2025-12-03T19:45:12Z",
"agent_type": "pydantic",
"error": "Connection timeout",
"duration_ms": 5000
}
],
"total": 1
}
"""
try:
ai_client = get_ai_client()
errors = await ai_client.get_recent_errors(limit=limit)
return {
"errors": errors,
"total": len(errors)
}
except Exception as e:
logger.error(f"Failed to fetch AI errors: {e}")
raise HTTPException(
status_code=503,
detail=f"Core-AI service unavailable: {str(e)}"
)
@router.get(
"/metrics/tool-failures",
response_model=Dict[str, Any],
summary="Get recent tool execution failures",
description="Returns recent tool execution failures with error details"
)
async def get_ai_tool_failures(limit: int = 20):
"""
Get recent tool execution failures
Args:
limit: Maximum number of failures to return (default: 20)
Returns:
Dict with failure list and total count
Example response:
{
"failures": [
{
"timestamp": "2025-12-03T19:50:30Z",
"tool_name": "list_containers",
"error": "Connection refused",
"duration_ms": 150
}
],
"total": 1
}
"""
try:
ai_client = get_ai_client()
failures = await ai_client.get_tool_failures(limit=limit)
return {
"failures": failures,
"total": len(failures)
}
except Exception as e:
logger.error(f"Failed to fetch tool failures: {e}")
raise HTTPException(
status_code=503,
detail=f"Core-AI service unavailable: {str(e)}"
)
@router.post(
"/metrics/reset",
summary="Reset all AI metrics (admin)",
description="Clear all collected metrics. This is an administrative operation that resets all counters and history."
)
async def reset_ai_metrics():
"""
Reset all AI metrics (admin operation)
Clears all collected metrics including:
- Request history
- Tool execution stats
- Memory system metrics
- Error logs
Returns:
Success confirmation
Note:
This is an administrative operation that should be used carefully.
All historical data will be lost.
"""
try:
ai_client = get_ai_client()
await ai_client.reset_metrics()
logger.info("AI metrics reset successfully")
return {
"success": True,
"message": "AI metrics reset successfully"
}
except Exception as e:
logger.error(f"Failed to reset AI metrics: {e}")
raise HTTPException(
status_code=503,
detail=f"Core-AI service unavailable: {str(e)}"
)
+23 -45
View File
@@ -10,10 +10,8 @@ from src.controllers.base import BaseController
from src.config import get_settings from src.config import get_settings
from src.logging_config import get_logger from src.logging_config import get_logger
from src.models.ollama_client import get_ollama_client from src.models.ollama_client import get_ollama_client
from src.db import get_database
# Note: Agent functionality moved to separate core-ai service (Dec 2025)
# This service (core-api) only provides infrastructure management and tools
AGENT_AVAILABLE = False
logger = get_logger(__name__) logger = get_logger(__name__)
@@ -52,20 +50,7 @@ class HealthController(BaseController):
"service": settings.app_name, "service": settings.app_name,
"version": settings.app_version, "version": settings.app_version,
"status": "healthy", "status": "healthy",
"documentation": { "docs": "/docs"
"swagger_ui": "/docs",
"redoc": "/redoc",
"openapi_spec": "/openapi.json"
},
"endpoints": {
"chat_completions": "/v1/chat/completions",
"models": "/v1/models",
"conversations": "/v1/conversations",
"dns_lookup": "/tools/dns/lookup",
"infrastructure": "/infrastructure",
"health": "/health",
"health_full": "/health/full"
}
} }
@router.get( @router.get(
@@ -75,18 +60,15 @@ class HealthController(BaseController):
) )
async def health_check(): async def health_check():
""" """
Simple health check endpoint for container orchestration Fast health check endpoint for container orchestration
Returns a 200 OK status when the service is running properly. Returns a 200 OK immediately if the service is running.
Used by Docker, Kubernetes, and load balancers. Does NOT check backend connectivity (use /health/full for that).
Used by Docker, Kubernetes, and load balancers for liveness probes.
""" """
ollama_client = get_ollama_client()
ollama_healthy = await ollama_client.health_check()
return { return {
"status": "healthy", "status": "healthy",
"version": settings.app_version, "version": settings.app_version
"ollama_connected": ollama_healthy
} }
@router.get( @router.get(
@@ -147,12 +129,18 @@ class HealthController(BaseController):
ollama_error = str(e) ollama_error = str(e)
logger.warning(f"Ollama health check failed: {ollama_error}") logger.warning(f"Ollama health check failed: {ollama_error}")
# Note: Agent functionality moved to separate core-ai service # Check 2: Database connection
# This service only needs Ollama for embeddings (infrastructure tools) database = get_database()
# Agent health is checked separately in core-ai service db_healthy = False
db_error = None
# Determine overall status (only Ollama required for core-api) try:
is_healthy = ollama_healthy db_healthy = await database.health_check()
except Exception as e:
db_error = str(e)
logger.warning(f"Database health check failed: {db_error}")
is_healthy = ollama_healthy and db_healthy
elapsed_ms = int((time.time() - start_time) * 1000) elapsed_ms = int((time.time() - start_time) * 1000)
status_code = 200 if is_healthy else 503 status_code = 200 if is_healthy else 503
@@ -171,7 +159,10 @@ class HealthController(BaseController):
}, },
"error": ollama_error "error": ollama_error
}, },
"note": "AI agent functionality available in separate core-ai service (port 8086)" "database": {
"status": "✅ healthy" if db_healthy else "❌ unhealthy",
"error": db_error
}
} }
} }
@@ -217,20 +208,7 @@ class HealthController(BaseController):
"error": str(e) "error": str(e)
} }
# 2. Agent Stack - Moved to separate core-ai service # 2. Configuration
diagnostics["components"]["agent"] = {
"status": "N/A",
"note": "AI agent functionality moved to separate core-ai service (port 8086)",
"check_url": "http://core-ai:8086/health"
}
# 3. Memory System (Qdrant) - Moved to core-ai service
diagnostics["components"]["qdrant"] = {
"status": "N/A",
"note": "Memory system managed by core-ai service (port 8086)"
}
# 4. Configuration
diagnostics["configuration"] = { diagnostics["configuration"] = {
"agent_fallback_enabled": settings.agent_fallback_enabled, "agent_fallback_enabled": settings.agent_fallback_enabled,
"memory_tier1_max_turns": settings.memory_tier1_max_turns, "memory_tier1_max_turns": settings.memory_tier1_max_turns,
@@ -4,6 +4,7 @@ Housekeeping Controller
Provides API endpoints for home automation via Home Assistant. Provides API endpoints for home automation via Home Assistant.
Designed for the Tatlock Housekeeper agent and other consumers. Designed for the Tatlock Housekeeper agent and other consumers.
""" """
import asyncio
from fastapi import APIRouter, HTTPException, Query, Depends from fastapi import APIRouter, HTTPException, Query, Depends
from typing import List, Dict, Any, Optional from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
@@ -413,6 +414,9 @@ class HousekeepingController(BaseController):
else: # toggle else: # toggle
await ha.toggle(entity_id) await ha.toggle(entity_id)
# Wait for HA to update state before fetching
await asyncio.sleep(0.3)
# Get new state # Get new state
new_state = await ha.get_state(entity_id) new_state = await ha.get_state(entity_id)
+353 -1
View File
@@ -4,7 +4,8 @@ Infrastructure Management Controller
Provides API endpoints for automated infrastructure management, Provides API endpoints for automated infrastructure management,
including service deployment, configuration, and monitoring setup. including service deployment, configuration, and monitoring setup.
""" """
from fastapi import APIRouter, HTTPException, Depends from fastapi import APIRouter, HTTPException, Depends, Request
from fastapi.responses import PlainTextResponse
from typing import List, Dict, Any, Optional, Union from typing import List, Dict, Any, Optional, Union
from pydantic import BaseModel, field_validator from pydantic import BaseModel, field_validator
@@ -1700,6 +1701,357 @@ class InfrastructureController(BaseController):
logger.error(f"Failed to get container resources: {e}", exc_info=True) logger.error(f"Failed to get container resources: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=str(e)) raise HTTPException(status_code=500, detail=str(e))
# ========================================================================
# Container Delete Endpoint (for Tatlock Control Room)
# ========================================================================
@router.delete(
"/containers/{container_id}",
status_code=204,
summary="Delete a container"
)
async def delete_container(
container_id: str,
force: bool = False
):
"""
Delete a Docker container.
Args:
container_id: Full container ID
force: Force remove running container (default: false)
Returns:
204 No Content on success
"""
portainer = get_portainer_client()
logger.info(f"Deleting container '{container_id}' (force={force})")
try:
# Get endpoint
endpoints = await portainer.get_endpoints()
if not endpoints:
raise HTTPException(status_code=500, detail="No Portainer endpoints available")
endpoint_id = endpoints[0]['Id']
# Delete the container
await portainer.delete_container(endpoint_id, container_id, force=force)
logger.info(f"Successfully deleted container '{container_id}'")
return None # 204 No Content
except HTTPException:
raise
except Exception as e:
error_str = str(e).lower()
if "404" in error_str or "no such container" in error_str:
raise HTTPException(status_code=404, detail=f"Container '{container_id}' not found")
elif "409" in error_str or "conflict" in error_str:
raise HTTPException(
status_code=409,
detail=f"Cannot delete container '{container_id}': container is running. Use force=true to force remove."
)
else:
logger.error(f"Failed to delete container '{container_id}': {e}", exc_info=True)
raise HTTPException(status_code=500, detail=str(e))
# ========================================================================
# Stack Management Endpoints (for Tatlock Control Room)
# ========================================================================
@router.get(
"/stacks/{stack_id}/compose",
summary="Get stack compose YAML",
response_class=PlainTextResponse
)
async def get_stack_compose(stack_id: str):
"""
Get the Docker Compose YAML for a stack.
Args:
stack_id: Stack name (compose project name)
Returns:
Docker Compose YAML as text/yaml
"""
portainer = get_portainer_client()
logger.info(f"Getting compose file for stack '{stack_id}'")
try:
# Find stack by name
stacks = await portainer.get_stacks()
stack = next(
(s for s in stacks if s.get("Name", "").lower() == stack_id.lower()),
None
)
if not stack:
raise HTTPException(status_code=404, detail=f"Stack '{stack_id}' not found")
# Get compose file content
compose_content = await portainer.get_stack_file(stack["Id"])
return PlainTextResponse(
content=compose_content,
media_type="text/yaml"
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to get compose for stack '{stack_id}': {e}", exc_info=True)
raise HTTPException(status_code=500, detail=str(e))
@router.put(
"/stacks/{stack_id}/compose",
status_code=204,
summary="Update stack compose YAML"
)
async def update_stack_compose(stack_id: str, request: Request):
"""
Update the Docker Compose YAML for a stack.
Args:
stack_id: Stack name (compose project name)
request: Raw YAML content in request body
Returns:
204 No Content on success
"""
portainer = get_portainer_client()
logger.info(f"Updating compose file for stack '{stack_id}'")
try:
# Read raw YAML from request body
compose_content = (await request.body()).decode("utf-8")
if not compose_content.strip():
raise HTTPException(status_code=400, detail="Empty compose content")
# Find stack by name
stacks = await portainer.get_stacks()
stack = next(
(s for s in stacks if s.get("Name", "").lower() == stack_id.lower()),
None
)
if not stack:
raise HTTPException(status_code=404, detail=f"Stack '{stack_id}' not found")
stack_int_id = stack["Id"]
endpoint_id = stack.get("EndpointId")
# Update stack with new compose content
await portainer.update_stack(
stack_id=stack_int_id,
stack_file_content=compose_content,
endpoint_id=endpoint_id,
prune=False,
pull_image=False
)
logger.info(f"Successfully updated compose for stack '{stack_id}'")
return None # 204 No Content
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to update compose for stack '{stack_id}': {e}", exc_info=True)
raise HTTPException(status_code=500, detail=str(e))
@router.get(
"/stacks/{stack_id}/env",
response_model=Dict[str, str],
summary="Get stack environment variables"
)
async def get_stack_env(stack_id: str):
"""
Get environment variables for a stack.
Args:
stack_id: Stack name (compose project name)
Returns:
Dictionary of environment variable name-value pairs
"""
portainer = get_portainer_client()
logger.info(f"Getting env vars for stack '{stack_id}'")
try:
# Find stack by name
stacks = await portainer.get_stacks()
stack = next(
(s for s in stacks if s.get("Name", "").lower() == stack_id.lower()),
None
)
if not stack:
raise HTTPException(status_code=404, detail=f"Stack '{stack_id}' not found")
# Get full stack details including env vars
stack_details = await portainer.get_stack(stack["Id"])
env_list = stack_details.get("Env", [])
# Convert from [{name, value}] to {name: value}
env_dict = {item["name"]: item["value"] for item in env_list}
return env_dict
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to get env for stack '{stack_id}': {e}", exc_info=True)
raise HTTPException(status_code=500, detail=str(e))
@router.put(
"/stacks/{stack_id}/env",
status_code=204,
summary="Update stack environment variables"
)
async def update_stack_env(stack_id: str, env_vars: Dict[str, str]):
"""
Update environment variables for a stack.
Args:
stack_id: Stack name (compose project name)
env_vars: Dictionary of environment variable name-value pairs
Returns:
204 No Content on success
"""
portainer = get_portainer_client()
logger.info(f"Updating env vars for stack '{stack_id}'")
try:
# Find stack by name
stacks = await portainer.get_stacks()
stack = next(
(s for s in stacks if s.get("Name", "").lower() == stack_id.lower()),
None
)
if not stack:
raise HTTPException(status_code=404, detail=f"Stack '{stack_id}' not found")
stack_int_id = stack["Id"]
endpoint_id = stack.get("EndpointId")
# Convert from {name: value} to [{name, value}]
env_list = [{"name": k, "value": v} for k, v in env_vars.items()]
# Update stack env vars
await portainer.update_stack_env(
stack_id=stack_int_id,
endpoint_id=endpoint_id,
env_vars=env_list
)
logger.info(f"Successfully updated env vars for stack '{stack_id}'")
return None # 204 No Content
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to update env for stack '{stack_id}': {e}", exc_info=True)
raise HTTPException(status_code=500, detail=str(e))
@router.post(
"/stacks/{stack_id}/deploy",
status_code=202,
summary="Deploy stack"
)
async def deploy_stack(stack_id: str):
"""
Redeploy a stack with current YAML and environment variables.
Args:
stack_id: Stack name (compose project name)
Returns:
202 Accepted
"""
portainer = get_portainer_client()
logger.info(f"Deploying stack '{stack_id}'")
try:
# Find stack by name
stacks = await portainer.get_stacks()
stack = next(
(s for s in stacks if s.get("Name", "").lower() == stack_id.lower()),
None
)
if not stack:
raise HTTPException(status_code=404, detail=f"Stack '{stack_id}' not found")
stack_int_id = stack["Id"]
endpoint_id = stack.get("EndpointId")
# Redeploy without pulling new images
await portainer.redeploy_stack(
stack_id=stack_int_id,
endpoint_id=endpoint_id,
pull_image=False
)
logger.info(f"Successfully deployed stack '{stack_id}'")
return None # 202 Accepted
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to deploy stack '{stack_id}': {e}", exc_info=True)
raise HTTPException(status_code=500, detail=str(e))
@router.post(
"/stacks/{stack_id}/rebuild",
status_code=202,
summary="Rebuild stack"
)
async def rebuild_stack(stack_id: str):
"""
Pull fresh images and recreate all containers in the stack.
Args:
stack_id: Stack name (compose project name)
Returns:
202 Accepted
"""
portainer = get_portainer_client()
logger.info(f"Rebuilding stack '{stack_id}'")
try:
# Find stack by name
stacks = await portainer.get_stacks()
stack = next(
(s for s in stacks if s.get("Name", "").lower() == stack_id.lower()),
None
)
if not stack:
raise HTTPException(status_code=404, detail=f"Stack '{stack_id}' not found")
stack_int_id = stack["Id"]
endpoint_id = stack.get("EndpointId")
# Redeploy with image pull
await portainer.redeploy_stack(
stack_id=stack_int_id,
endpoint_id=endpoint_id,
pull_image=True
)
logger.info(f"Successfully rebuilt stack '{stack_id}'")
return None # 202 Accepted
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to rebuild stack '{stack_id}': {e}", exc_info=True)
raise HTTPException(status_code=500, detail=str(e))
return router return router
-23
View File
@@ -1,23 +0,0 @@
"""
Infrastructure Credentials Template
INSTRUCTIONS:
1. Copy this file to credentials.py
2. Fill in your actual credentials
3. DO NOT commit credentials.py to version control (it's in .gitignore)
This file should be committed to the repository as a template.
"""
# Portainer Configuration
PORTAINER_URL = "http://localhost:8001"
PORTAINER_API_KEY = "ptr_your_api_token_here" # Create in Portainer UI: User menu → My account → Access tokens
# Nginx Proxy Manager Configuration
NPM_URL = "http://localhost:81"
NPM_EMAIL = "admin@example.com"
NPM_PASSWORD = "your_password_here"
# Home Assistant Configuration
HOMEASSISTANT_URL = "http://192.168.86.149:8123" # Or http://home-assistant:8123 in Docker
HOMEASSISTANT_TOKEN = "your_long_lived_access_token_here" # Create in HA: Profile → Long-Lived Access Tokens
+16
View File
@@ -0,0 +1,16 @@
"""
Database package for Core-API
Provides async PostgreSQL database connectivity using SQLAlchemy 2.0.
"""
from src.db.database import (
get_async_session,
get_database,
Database,
)
__all__ = [
"get_async_session",
"get_database",
"Database",
]
+179
View File
@@ -0,0 +1,179 @@
"""
Database Connection Module
Provides async PostgreSQL connectivity using SQLAlchemy 2.0 with asyncpg driver.
Follows the existing singleton pattern used throughout core-api.
"""
from typing import AsyncGenerator, Optional
from sqlalchemy import text
from sqlalchemy.ext.asyncio import (
AsyncSession,
AsyncEngine,
create_async_engine,
async_sessionmaker,
)
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.pool import NullPool
from src.config import get_settings
from src.logging_config import get_logger
logger = get_logger(__name__)
settings = get_settings()
class Base(DeclarativeBase):
"""
SQLAlchemy declarative base for all models
All database models should inherit from this class.
"""
pass
class Database:
"""
Async database connection manager
Provides async engine and session factory for PostgreSQL connections.
Uses asyncpg driver for optimal async performance.
"""
def __init__(self, database_url: Optional[str] = None):
"""
Initialize database connection manager
Args:
database_url: PostgreSQL connection URL (default from settings)
"""
# Convert postgresql:// to postgresql+asyncpg:// for async driver
url = database_url or settings.database_url
if url.startswith("postgresql://"):
url = url.replace("postgresql://", "postgresql+asyncpg://", 1)
self._url = url
self._engine: Optional[AsyncEngine] = None
self._session_factory: Optional[async_sessionmaker[AsyncSession]] = None
@property
def engine(self) -> AsyncEngine:
"""
Get or create the async database engine
Returns:
AsyncEngine instance
"""
if self._engine is None:
self._engine = create_async_engine(
self._url,
echo=settings.debug, # Log SQL in debug mode
poolclass=NullPool, # Disable connection pooling for serverless compatibility
)
logger.info(f"Database engine created for {self._url.split('@')[-1]}")
return self._engine
@property
def session_factory(self) -> async_sessionmaker[AsyncSession]:
"""
Get or create the async session factory
Returns:
Session factory for creating database sessions
"""
if self._session_factory is None:
self._session_factory = async_sessionmaker(
bind=self.engine,
class_=AsyncSession,
expire_on_commit=False,
autocommit=False,
autoflush=False,
)
return self._session_factory
async def create_tables(self) -> None:
"""
Create all database tables
Should only be used for development/testing.
Use Alembic migrations for production.
"""
async with self.engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
logger.info("Database tables created")
async def drop_tables(self) -> None:
"""
Drop all database tables
WARNING: Destroys all data. Use with caution.
"""
async with self.engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
logger.warning("Database tables dropped")
async def health_check(self) -> bool:
"""
Check if database connection is healthy
Returns:
True if connection successful, False otherwise
"""
try:
async with self.session_factory() as session:
await session.execute(text("SELECT 1"))
return True
except Exception as e:
logger.error(f"Database health check failed: {e}")
return False
async def close(self) -> None:
"""
Close database connections and dispose of engine
"""
if self._engine is not None:
await self._engine.dispose()
self._engine = None
self._session_factory = None
logger.info("Database connections closed")
# Singleton instance
_database: Optional[Database] = None
def get_database() -> Database:
"""
Get singleton database instance
Returns:
Database instance
"""
global _database
if _database is None:
_database = Database()
return _database
async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
"""
FastAPI dependency for database sessions
Yields an async session that is automatically closed after the request.
Usage:
@router.get("/items")
async def get_items(session: AsyncSession = Depends(get_async_session)):
result = await session.execute(select(Item))
return result.scalars().all()
Yields:
AsyncSession instance
"""
database = get_database()
async with database.session_factory() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
+17
View File
@@ -0,0 +1,17 @@
"""
SQLAlchemy Models for Core-API
Database models for authentication, authorization, and user management.
"""
from src.db.models.user import User
from src.db.models.role import Role, UserRole
from src.db.models.user_preferences import UserPreferences
from src.db.models.api_key import ApiKey
__all__ = [
"User",
"Role",
"UserRole",
"UserPreferences",
"ApiKey",
]
+98
View File
@@ -0,0 +1,98 @@
"""
API Key Model
Provides API key authentication as fallback for OIDC.
Keys are tied to user accounts and inherit user permissions.
"""
import uuid
from datetime import datetime
from sqlalchemy import String, ForeignKey, DateTime, func
from sqlalchemy.dialects.postgresql import UUID, ARRAY
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src.db.database import Base
from typing import TYPE_CHECKING, List
if TYPE_CHECKING:
from src.db.models.user import User
class ApiKey(Base):
"""
API Key model for programmatic access
API keys provide an alternative to OIDC for:
- Local development without SSO
- Service-to-service communication
- Scripts and automation
Keys inherit the user's roles but can optionally
be restricted to a subset of scopes.
"""
__tablename__ = "api_keys"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
primary_key=True,
default=uuid.uuid4,
)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
name: Mapped[str] = mapped_column(
String(100),
nullable=False,
comment="Human-readable key name (e.g., 'Dev Laptop', 'CI/CD')",
)
key_hash: Mapped[str] = mapped_column(
String(255),
nullable=False,
comment="SHA-256 hash of the API key",
)
key_prefix: Mapped[str] = mapped_column(
String(8),
nullable=False,
comment="First 8 chars of key for identification (e.g., 'cak_abc1')",
)
scopes: Mapped[List[str] | None] = mapped_column(
ARRAY(String),
nullable=True,
comment="Optional scope restriction (subset of user roles)",
)
expires_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
comment="Optional expiration timestamp",
)
last_used_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
comment="Last time this key was used",
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
nullable=False,
)
# Relationships
user: Mapped["User"] = relationship(
"User",
back_populates="api_keys",
)
def __repr__(self) -> str:
return f"<ApiKey {self.key_prefix}... ({self.name})>"
@property
def is_expired(self) -> bool:
"""Check if the API key has expired"""
if self.expires_at is None:
return False
return datetime.now(self.expires_at.tzinfo) > self.expires_at
+93
View File
@@ -0,0 +1,93 @@
"""
Role Models
Defines domain-scoped permissions mapped from Authentik groups.
Format: {domain}:{action} (e.g., control-room:admin, media:viewer)
"""
import uuid
from typing import TYPE_CHECKING, List
from sqlalchemy import String, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src.db.database import Base
if TYPE_CHECKING:
from src.db.models.user import User
class Role(Base):
"""
Role model for domain-scoped permissions
Roles are seeded from configuration, not user-editable.
Each role maps to an Authentik group (e.g., tatlock-control-room-admin).
Domains: control-room, library, media, ai, housekeeper, developer, documents, gaming, admin
Actions: viewer, user, editor, admin (hierarchical)
"""
__tablename__ = "roles"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
primary_key=True,
default=uuid.uuid4,
)
name: Mapped[str] = mapped_column(
String(100),
unique=True,
nullable=False,
index=True,
comment="Role name in format domain:action (e.g., control-room:admin)",
)
domain: Mapped[str] = mapped_column(
String(50),
nullable=False,
index=True,
comment="Permission domain (e.g., control-room, media, ai)",
)
action: Mapped[str] = mapped_column(
String(20),
nullable=False,
comment="Permission action (viewer, user, editor, admin)",
)
authentik_group: Mapped[str | None] = mapped_column(
String(255),
nullable=True,
unique=True,
comment="Corresponding Authentik group name (e.g., tatlock-control-room-admin)",
)
# Relationships
users: Mapped[List["User"]] = relationship(
"User",
secondary="user_roles",
back_populates="roles",
lazy="selectin",
)
def __repr__(self) -> str:
return f"<Role {self.name}>"
class UserRole(Base):
"""
Association table for User-Role many-to-many relationship
Synced from Authentik groups during user authentication.
"""
__tablename__ = "user_roles"
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("users.id", ondelete="CASCADE"),
primary_key=True,
)
role_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("roles.id", ondelete="CASCADE"),
primary_key=True,
)
+94
View File
@@ -0,0 +1,94 @@
"""
User Model
Represents users synced from Authentik SSO.
"""
import uuid
from datetime import datetime
from typing import TYPE_CHECKING, List
from sqlalchemy import String, Boolean, DateTime, func
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src.db.database import Base
if TYPE_CHECKING:
from src.db.models.role import Role
from src.db.models.user_preferences import UserPreferences
from src.db.models.api_key import ApiKey
class User(Base):
"""
User model synced from Authentik
Users are created/updated when they authenticate via OIDC.
The authentik_id links to the Authentik user record.
"""
__tablename__ = "users"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
primary_key=True,
default=uuid.uuid4,
)
authentik_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
unique=True,
nullable=False,
index=True,
)
email: Mapped[str] = mapped_column(
String(255),
unique=True,
nullable=False,
index=True,
)
name: Mapped[str] = mapped_column(
String(255),
nullable=False,
)
avatar_url: Mapped[str | None] = mapped_column(
String(500),
nullable=True,
)
api_keys_enabled: Mapped[bool] = mapped_column(
Boolean,
default=True,
nullable=False,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
nullable=False,
)
last_login: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
# Relationships
roles: Mapped[List["Role"]] = relationship(
"Role",
secondary="user_roles",
back_populates="users",
lazy="selectin",
)
preferences: Mapped["UserPreferences"] = relationship(
"UserPreferences",
back_populates="user",
uselist=False,
lazy="selectin",
cascade="all, delete-orphan",
)
api_keys: Mapped[List["ApiKey"]] = relationship(
"ApiKey",
back_populates="user",
lazy="selectin",
cascade="all, delete-orphan",
)
def __repr__(self) -> str:
return f"<User {self.email}>"
+61
View File
@@ -0,0 +1,61 @@
"""
User Preferences Model
Stores user-specific settings like theme and default room.
"""
import uuid
from sqlalchemy import String, ForeignKey
from sqlalchemy.dialects.postgresql import UUID, JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src.db.database import Base
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from src.db.models.user import User
class UserPreferences(Base):
"""
User preferences model
Stores user-specific settings that persist across sessions.
Extended settings stored in preferences_json for flexibility.
"""
__tablename__ = "user_preferences"
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("users.id", ondelete="CASCADE"),
primary_key=True,
)
theme: Mapped[str] = mapped_column(
String(20),
default="system",
nullable=False,
comment="Theme preference: system, light, dark",
)
default_room: Mapped[str] = mapped_column(
String(50),
default="front-hall",
nullable=False,
comment="Default room for housekeeping features",
)
preferences_json: Mapped[dict] = mapped_column(
JSONB,
default=dict,
nullable=False,
comment="Extended preferences as JSON",
)
# Relationships
user: Mapped["User"] = relationship(
"User",
back_populates="preferences",
)
def __repr__(self) -> str:
return f"<UserPreferences user_id={self.user_id}>"
+19 -80
View File
@@ -9,12 +9,13 @@ from contextlib import asynccontextmanager
from src.config import get_settings from src.config import get_settings
from src.logging_config import setup_logging, get_logger from src.logging_config import setup_logging, get_logger
from src.models.ollama_client import get_ollama_client, close_ollama_client from src.models.ollama_client import get_ollama_client, close_ollama_client
from src.db import get_database
from src.controllers.infrastructure_controller import infrastructure_controller from src.controllers.infrastructure_controller import infrastructure_controller
from src.controllers.tools_controller import tools_controller from src.controllers.tools_controller import tools_controller
from src.controllers.health_controller import health_controller from src.controllers.health_controller import health_controller
from src.controllers.static_controller import static_controller from src.controllers.static_controller import static_controller
from src.controllers.ai_controller import router as ai_router
from src.controllers.housekeeping_controller import housekeeping_controller from src.controllers.housekeeping_controller import housekeeping_controller
from src.auth.controller import auth_controller
from src.security import initialize_oidc from src.security import initialize_oidc
# Initialize settings # Initialize settings
@@ -49,6 +50,14 @@ async def lifespan(app: FastAPI):
else: else:
logger.warning("✗ Ollama connection failed - AI features may not work") logger.warning("✗ Ollama connection failed - AI features may not work")
# Check database connectivity
database = get_database()
db_healthy = await database.health_check()
if db_healthy:
logger.info("✓ Database connection successful")
else:
logger.warning("✗ Database connection failed - auth features may not work")
# Initialize security (OIDC authentication) # Initialize security (OIDC authentication)
initialize_oidc(settings) initialize_oidc(settings)
@@ -57,6 +66,7 @@ async def lifespan(app: FastAPI):
# Shutdown # Shutdown
logger.info("Shutting down application") logger.info("Shutting down application")
await close_ollama_client() await close_ollama_client()
await database.close()
# Create FastAPI application # Create FastAPI application
@@ -64,89 +74,18 @@ app = FastAPI(
title=settings.app_name, title=settings.app_name,
version=settings.app_version, version=settings.app_version,
description=""" description="""
Core Code API provides OpenAPI-compatible functions and AI orchestration for Open WebUI. Core Code API - Infrastructure management and home automation API.
## Features ## Features
### OpenAI-Compatible API (v1) - **Infrastructure Management** - Container and stack management via Portainer
- `/v1/chat/completions` - Chat completions with streaming support - **Home Automation** - Device control via Home Assistant
- `/v1/models` - List available models - **Tools** - DNS lookup and utilities
Compatible with OpenAI client libraries and Open WebUI.
### Conversation Memory (Phase 2) See `/docs` for the full API reference.
- `/v1/conversations/{id}` - Get conversation history
- `/v1/conversations/{id}/search` - Semantic search within conversation
- `/v1/conversations/search` - Search across all conversations
- `/v1/conversations/{id}/stats` - Get conversation statistics
- `/v1/conversations/{id}/consolidate` - Manual consolidation
- `DELETE /v1/conversations/{id}` - Delete conversation
Multi-tier memory system:
- **Tier 1**: Fast in-memory buffer (last 10 turns)
- **Tier 2/3**: Unified Qdrant storage (persistent + semantic search)
### Infrastructure Management
**Read Endpoints:**
- `GET /infrastructure/health` - Check Portainer & NPM connectivity
- `GET /infrastructure/services` - List all deployed services
- `GET /infrastructure/services/{name}` - Get service details
- `GET /infrastructure/ports` - List allocated ports
- `GET /infrastructure/domains` - List configured domains
**Write Endpoints (Admin Only):**
- `POST /infrastructure/services` - Deploy new service from compose YAML
- `PUT /infrastructure/services/{name}` - Update existing service
- `DELETE /infrastructure/services/{name}` - Remove service and stack
- `POST /infrastructure/proxy` - Create proxy host with optional SSL
Automates infrastructure operations via Portainer and Nginx Proxy Manager APIs.
### Home Automation (Housekeeping)
**Read Endpoints:**
- `GET /housekeeping/health` - Check Home Assistant connectivity
- `GET /housekeeping/devices` - List devices (filter by domain, area)
- `GET /housekeeping/devices/{entity_id}` - Get device details
- `GET /housekeeping/areas` - List areas/rooms
- `GET /housekeeping/scenes` - List scenes
- `GET /housekeeping/scripts` - List scripts
- `GET /housekeeping/automations` - List automations
- `GET /housekeeping/history` - Get state history
**Write Endpoints (Admin Only):**
- `POST /housekeeping/devices/{entity_id}/control` - Control device
- `POST /housekeeping/scenes/{scene_id}/activate` - Activate scene
- `POST /housekeeping/scripts/{script_id}/run` - Run script
- `POST /housekeeping/automations/{automation_id}/toggle` - Enable/disable automation
Abstracts Home Assistant for the Tatlock Housekeeper agent and other consumers.
### Web Scraper
Intelligent web scraping with main content extraction.
Perfect for extracting articles, documentation, and blog posts for LLM consumption.
## Authentication
When OIDC authentication is enabled (oidc_enabled=true in config):
- Infrastructure write endpoints require authentication
- Use OAuth2/OIDC bearer token from Authentik
- Admin group membership required for infrastructure operations
## Integration
This API is designed to integrate with:
- **Open WebUI**: Direct OpenAI API compatibility
- **Open WebUI Functions**: Import via OpenAPI spec
- **Open WebUI Pipelines**: Use as data source
- **LangChain**: Compatible with standard HTTP tools
## Documentation
- **OpenAPI Spec**: `/openapi.json`
- **Swagger UI**: `/docs`
- **ReDoc**: `/redoc`
""", """,
docs_url="/docs", docs_url="/docs",
redoc_url="/redoc", redoc_url=None,
openapi_url="/openapi.json", openapi_url="/openapi.json",
lifespan=lifespan, lifespan=lifespan,
debug=settings.debug, debug=settings.debug,
@@ -168,11 +107,11 @@ app.add_middleware(
# Include controller routers # Include controller routers
app.include_router(health_controller.router) # / and /health app.include_router(health_controller.router) # / and /health
app.include_router(auth_controller.router) # /auth/*
app.include_router(tools_controller.router) # /tools/* app.include_router(tools_controller.router) # /tools/*
app.include_router(infrastructure_controller.router) # /infrastructure/* app.include_router(infrastructure_controller.router) # /infrastructure/*
app.include_router(housekeeping_controller.router) # /housekeeping/* app.include_router(housekeeping_controller.router) # /housekeeping/*
app.include_router(static_controller.router) # /static/* app.include_router(static_controller.router) # /static/*
app.include_router(ai_router) # /ai/*
# Global exception handler # Global exception handler
+376
View File
@@ -402,3 +402,379 @@ class TestInfrastructureGetService:
response = client.get("/infrastructure/services/nonexistent") response = client.get("/infrastructure/services/nonexistent")
assert response.status_code == 404 assert response.status_code == 404
class TestInfrastructureDeleteContainer:
"""Test DELETE /infrastructure/containers/{container_id} endpoint."""
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_delete_container_returns_204(self, mock_get_npm, mock_get_portainer, client):
"""Delete container should return 204 on success."""
mock_portainer = AsyncMock()
mock_portainer.get_endpoints.return_value = [{"Id": 1}]
mock_portainer.delete_container.return_value = True
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.delete("/infrastructure/containers/abc123")
assert response.status_code == 204
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_delete_container_with_force(self, mock_get_npm, mock_get_portainer, client):
"""Delete container should pass force parameter."""
mock_portainer = AsyncMock()
mock_portainer.get_endpoints.return_value = [{"Id": 1}]
mock_portainer.delete_container.return_value = True
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.delete("/infrastructure/containers/abc123?force=true")
assert response.status_code == 204
mock_portainer.delete_container.assert_called_once_with(1, "abc123", force=True)
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_delete_container_returns_404_for_missing(self, mock_get_npm, mock_get_portainer, client):
"""Delete container should return 404 if not found."""
mock_portainer = AsyncMock()
mock_portainer.get_endpoints.return_value = [{"Id": 1}]
mock_portainer.delete_container.side_effect = Exception("404 no such container")
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.delete("/infrastructure/containers/nonexistent")
assert response.status_code == 404
class TestInfrastructureStackCompose:
"""Test /infrastructure/stacks/{stackId}/compose endpoints."""
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_get_stack_compose_returns_yaml(self, mock_get_npm, mock_get_portainer, client):
"""Get stack compose should return YAML content."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = [
{"Id": 1, "Name": "mystack", "EndpointId": 1}
]
mock_portainer.get_stack_file.return_value = "version: '3'\nservices:\n web:\n image: nginx"
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.get("/infrastructure/stacks/mystack/compose")
assert response.status_code == 200
assert "text/yaml" in response.headers.get("content-type", "")
assert "version:" in response.text
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_get_stack_compose_returns_404_for_missing(self, mock_get_npm, mock_get_portainer, client):
"""Get stack compose should return 404 if stack not found."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = []
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.get("/infrastructure/stacks/nonexistent/compose")
assert response.status_code == 404
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_put_stack_compose_returns_204(self, mock_get_npm, mock_get_portainer, client):
"""Update stack compose should return 204 on success."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = [
{"Id": 1, "Name": "mystack", "EndpointId": 1}
]
mock_portainer.update_stack.return_value = {"Id": 1}
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.put(
"/infrastructure/stacks/mystack/compose",
content="version: '3'\nservices:\n web:\n image: nginx:latest",
headers={"Content-Type": "text/yaml"}
)
assert response.status_code == 204
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_put_stack_compose_returns_400_for_empty(self, mock_get_npm, mock_get_portainer, client):
"""Update stack compose should return 400 for empty content."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = [
{"Id": 1, "Name": "mystack", "EndpointId": 1}
]
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.put(
"/infrastructure/stacks/mystack/compose",
content="",
headers={"Content-Type": "text/yaml"}
)
assert response.status_code == 400
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_put_stack_compose_returns_404_for_missing(self, mock_get_npm, mock_get_portainer, client):
"""Update stack compose should return 404 if stack not found."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = []
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.put(
"/infrastructure/stacks/nonexistent/compose",
content="version: '3'",
headers={"Content-Type": "text/yaml"}
)
assert response.status_code == 404
class TestInfrastructureStackEnv:
"""Test /infrastructure/stacks/{stackId}/env endpoints."""
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_get_stack_env_returns_dict(self, mock_get_npm, mock_get_portainer, client):
"""Get stack env should return environment variables as dict."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = [
{"Id": 1, "Name": "mystack", "EndpointId": 1}
]
mock_portainer.get_stack.return_value = {
"Id": 1,
"Name": "mystack",
"Env": [
{"name": "DB_HOST", "value": "localhost"},
{"name": "DB_PORT", "value": "5432"}
]
}
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.get("/infrastructure/stacks/mystack/env")
assert response.status_code == 200
data = response.json()
assert data == {"DB_HOST": "localhost", "DB_PORT": "5432"}
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_get_stack_env_returns_empty_dict(self, mock_get_npm, mock_get_portainer, client):
"""Get stack env should return empty dict if no env vars."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = [
{"Id": 1, "Name": "mystack", "EndpointId": 1}
]
mock_portainer.get_stack.return_value = {"Id": 1, "Name": "mystack", "Env": []}
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.get("/infrastructure/stacks/mystack/env")
assert response.status_code == 200
assert response.json() == {}
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_get_stack_env_returns_404_for_missing(self, mock_get_npm, mock_get_portainer, client):
"""Get stack env should return 404 if stack not found."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = []
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.get("/infrastructure/stacks/nonexistent/env")
assert response.status_code == 404
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_put_stack_env_returns_204(self, mock_get_npm, mock_get_portainer, client):
"""Update stack env should return 204 on success."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = [
{"Id": 1, "Name": "mystack", "EndpointId": 1}
]
mock_portainer.update_stack_env.return_value = {"Id": 1}
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.put(
"/infrastructure/stacks/mystack/env",
json={"DB_HOST": "newhost", "DB_PORT": "5433"}
)
assert response.status_code == 204
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_put_stack_env_returns_404_for_missing(self, mock_get_npm, mock_get_portainer, client):
"""Update stack env should return 404 if stack not found."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = []
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.put(
"/infrastructure/stacks/nonexistent/env",
json={"KEY": "value"}
)
assert response.status_code == 404
class TestInfrastructureStackDeploy:
"""Test /infrastructure/stacks/{stackId}/deploy endpoint."""
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_deploy_stack_returns_202(self, mock_get_npm, mock_get_portainer, client):
"""Deploy stack should return 202 Accepted."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = [
{"Id": 1, "Name": "mystack", "EndpointId": 1}
]
mock_portainer.redeploy_stack.return_value = {"Id": 1}
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.post("/infrastructure/stacks/mystack/deploy")
assert response.status_code == 202
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_deploy_stack_does_not_pull_images(self, mock_get_npm, mock_get_portainer, client):
"""Deploy stack should not pull images."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = [
{"Id": 1, "Name": "mystack", "EndpointId": 1}
]
mock_portainer.redeploy_stack.return_value = {"Id": 1}
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.post("/infrastructure/stacks/mystack/deploy")
assert response.status_code == 202
mock_portainer.redeploy_stack.assert_called_once_with(
stack_id=1,
endpoint_id=1,
pull_image=False
)
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_deploy_stack_returns_404_for_missing(self, mock_get_npm, mock_get_portainer, client):
"""Deploy stack should return 404 if stack not found."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = []
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.post("/infrastructure/stacks/nonexistent/deploy")
assert response.status_code == 404
class TestInfrastructureStackRebuild:
"""Test /infrastructure/stacks/{stackId}/rebuild endpoint."""
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_rebuild_stack_returns_202(self, mock_get_npm, mock_get_portainer, client):
"""Rebuild stack should return 202 Accepted."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = [
{"Id": 1, "Name": "mystack", "EndpointId": 1}
]
mock_portainer.redeploy_stack.return_value = {"Id": 1}
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.post("/infrastructure/stacks/mystack/rebuild")
assert response.status_code == 202
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_rebuild_stack_pulls_images(self, mock_get_npm, mock_get_portainer, client):
"""Rebuild stack should pull images."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = [
{"Id": 1, "Name": "mystack", "EndpointId": 1}
]
mock_portainer.redeploy_stack.return_value = {"Id": 1}
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.post("/infrastructure/stacks/mystack/rebuild")
assert response.status_code == 202
mock_portainer.redeploy_stack.assert_called_once_with(
stack_id=1,
endpoint_id=1,
pull_image=True
)
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_rebuild_stack_returns_404_for_missing(self, mock_get_npm, mock_get_portainer, client):
"""Rebuild stack should return 404 if stack not found."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = []
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.post("/infrastructure/stacks/nonexistent/rebuild")
assert response.status_code == 404
@patch("src.controllers.infrastructure_controller.get_portainer_client")
@patch("src.controllers.infrastructure_controller.get_npm_client")
def test_rebuild_stack_case_insensitive(self, mock_get_npm, mock_get_portainer, client):
"""Rebuild stack should match stack name case-insensitively."""
mock_portainer = AsyncMock()
mock_portainer.get_stacks.return_value = [
{"Id": 1, "Name": "MyStack", "EndpointId": 1}
]
mock_portainer.redeploy_stack.return_value = {"Id": 1}
mock_get_portainer.return_value = mock_portainer
mock_npm = AsyncMock()
mock_get_npm.return_value = mock_npm
response = client.post("/infrastructure/stacks/mystack/rebuild")
assert response.status_code == 202
+233 -112
View File
@@ -391,40 +391,12 @@ class TestPortainerClientContainers:
assert result is True assert result is True
class TestPortainerClientDockerSocketFallback:
"""Test Docker socket fallback methods."""
@pytest.mark.asyncio
async def test_list_containers_via_socket_returns_empty_on_error(self):
"""_list_containers_via_socket should return empty list on error."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
with patch("httpx.AsyncHTTPTransport") as mock_transport:
mock_transport.side_effect = Exception("Socket not available")
result = await client._list_containers_via_socket()
assert result == []
@pytest.mark.asyncio
async def test_inspect_container_via_socket_returns_none_on_error(self):
"""_inspect_container_via_socket should return None on error."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
with patch("httpx.AsyncHTTPTransport") as mock_transport:
mock_transport.side_effect = Exception("Socket not available")
result = await client._inspect_container_via_socket("container_name")
assert result is None
class TestPortainerClientWrapperMethods: class TestPortainerClientWrapperMethods:
"""Test convenience wrapper methods.""" """Test convenience wrapper methods."""
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_list_containers_uses_portainer_first(self): async def test_list_containers_uses_portainer(self):
"""list_containers should try Portainer first.""" """list_containers should use Portainer API."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key") client = PortainerClient(base_url="http://portainer:9000", api_key="key")
containers = [{"Id": "abc123", "Names": ["/test"]}] containers = [{"Id": "abc123", "Names": ["/test"]}]
@@ -442,57 +414,19 @@ class TestPortainerClientWrapperMethods:
mock_containers.assert_called_once() mock_containers.assert_called_once()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_list_containers_falls_back_to_socket(self): async def test_list_containers_raises_when_no_endpoints(self):
"""list_containers should fallback to Docker socket if Portainer returns empty.""" """list_containers should raise RuntimeError when no endpoints available."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key") client = PortainerClient(base_url="http://portainer:9000", api_key="key")
with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints: with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints:
mock_endpoints.return_value = [{"Id": 1}] mock_endpoints.return_value = []
with patch.object(client, "get_containers", new_callable=AsyncMock) as mock_containers: with pytest.raises(RuntimeError, match="No Portainer endpoints available"):
mock_containers.return_value = [] await client.list_containers()
with patch.object(client, "_list_containers_via_socket", new_callable=AsyncMock) as mock_socket:
mock_socket.return_value = [{"Id": "from_socket"}]
result = await client.list_containers()
assert result == [{"Id": "from_socket"}]
mock_socket.assert_called_once()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_list_containers_handles_exception_with_fallback(self): async def test_inspect_container_uses_portainer(self):
"""list_containers should try fallback even on exception.""" """inspect_container should use Portainer API."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints:
mock_endpoints.side_effect = Exception("API error")
with patch.object(client, "_list_containers_via_socket", new_callable=AsyncMock) as mock_socket:
mock_socket.return_value = [{"Id": "fallback"}]
result = await client.list_containers()
assert result == [{"Id": "fallback"}]
@pytest.mark.asyncio
async def test_list_containers_returns_empty_when_all_fails(self):
"""list_containers should return empty list when everything fails."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints:
mock_endpoints.side_effect = Exception("API error")
with patch.object(client, "_list_containers_via_socket", new_callable=AsyncMock) as mock_socket:
mock_socket.side_effect = Exception("Socket error")
result = await client.list_containers()
assert result == []
@pytest.mark.asyncio
async def test_inspect_container_uses_portainer_first(self):
"""inspect_container should try Portainer first."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key") client = PortainerClient(base_url="http://portainer:9000", api_key="key")
container_list = [{"Id": "abc123", "Names": ["/mycontainer"]}] container_list = [{"Id": "abc123", "Names": ["/mycontainer"]}]
@@ -512,54 +446,241 @@ class TestPortainerClientWrapperMethods:
assert result == container_detail assert result == container_detail
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_inspect_container_falls_back_to_socket(self): async def test_inspect_container_returns_none_when_not_found(self):
"""inspect_container should fallback if not found in Portainer.""" """inspect_container should return None if container not found."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key") client = PortainerClient(base_url="http://portainer:9000", api_key="key")
with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints: with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints:
mock_endpoints.return_value = [{"Id": 1}] mock_endpoints.return_value = [{"Id": 1}]
with patch.object(client, "get_containers", new_callable=AsyncMock) as mock_list: with patch.object(client, "get_containers", new_callable=AsyncMock) as mock_list:
mock_list.return_value = [] # Container not found mock_list.return_value = [] # No containers
with patch.object(client, "_inspect_container_via_socket", new_callable=AsyncMock) as mock_socket:
mock_socket.return_value = {"Id": "from_socket"}
result = await client.inspect_container("missing_container") result = await client.inspect_container("missing_container")
assert result == {"Id": "from_socket"}
mock_socket.assert_called_once()
@pytest.mark.asyncio
async def test_inspect_container_handles_exception_with_fallback(self):
"""inspect_container should try fallback even on exception."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints:
mock_endpoints.side_effect = Exception("API error")
with patch.object(client, "_inspect_container_via_socket", new_callable=AsyncMock) as mock_socket:
mock_socket.return_value = {"Id": "fallback"}
result = await client.inspect_container("container")
assert result == {"Id": "fallback"}
@pytest.mark.asyncio
async def test_inspect_container_returns_none_when_all_fails(self):
"""inspect_container should return None when everything fails."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints:
mock_endpoints.side_effect = Exception("API error")
with patch.object(client, "_inspect_container_via_socket", new_callable=AsyncMock) as mock_socket:
mock_socket.side_effect = Exception("Socket error")
result = await client.inspect_container("container")
assert result is None assert result is None
@pytest.mark.asyncio
async def test_inspect_container_raises_when_no_endpoints(self):
"""inspect_container should raise RuntimeError when no endpoints available."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
with patch.object(client, "get_endpoints", new_callable=AsyncMock) as mock_endpoints:
mock_endpoints.return_value = []
with pytest.raises(RuntimeError, match="No Portainer endpoints available"):
await client.inspect_container("container")
class TestPortainerClientStackFile:
"""Test stack file operations."""
@pytest.mark.asyncio
async def test_get_stack_file_returns_content(self):
"""get_stack_file should return compose YAML content."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
compose_content = "version: '3'\nservices:\n web:\n image: nginx"
with patch("httpx.AsyncClient") as mock_client_class:
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"StackFileContent": compose_content}
mock_response.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.get.return_value = mock_response
mock_client.__aenter__.return_value = mock_client
mock_client.__aexit__.return_value = None
mock_client_class.return_value = mock_client
result = await client.get_stack_file(1)
assert result == compose_content
@pytest.mark.asyncio
async def test_get_stack_file_returns_empty_string_if_missing(self):
"""get_stack_file should return empty string if content missing."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
with patch("httpx.AsyncClient") as mock_client_class:
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {}
mock_response.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.get.return_value = mock_response
mock_client.__aenter__.return_value = mock_client
mock_client.__aexit__.return_value = None
mock_client_class.return_value = mock_client
result = await client.get_stack_file(1)
assert result == ""
class TestPortainerClientRedeployStack:
"""Test stack redeploy operations."""
@pytest.mark.asyncio
async def test_redeploy_stack_without_pull(self):
"""redeploy_stack should redeploy without pulling images by default."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
with patch.object(client, "get_stack_file", new_callable=AsyncMock) as mock_file:
mock_file.return_value = "version: '3'"
with patch.object(client, "get_stack", new_callable=AsyncMock) as mock_stack:
mock_stack.return_value = {"Id": 1, "Env": [{"name": "KEY", "value": "val"}]}
with patch("httpx.AsyncClient") as mock_client_class:
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"Id": 1}
mock_response.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.put.return_value = mock_response
mock_client.__aenter__.return_value = mock_client
mock_client.__aexit__.return_value = None
mock_client_class.return_value = mock_client
result = await client.redeploy_stack(1, 1, pull_image=False)
call_args = mock_client.put.call_args
assert call_args[1]["json"]["pullImage"] is False
assert result == {"Id": 1}
@pytest.mark.asyncio
async def test_redeploy_stack_with_pull(self):
"""redeploy_stack should pull images when requested."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
with patch.object(client, "get_stack_file", new_callable=AsyncMock) as mock_file:
mock_file.return_value = "version: '3'"
with patch.object(client, "get_stack", new_callable=AsyncMock) as mock_stack:
mock_stack.return_value = {"Id": 1, "Env": []}
with patch("httpx.AsyncClient") as mock_client_class:
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"Id": 1}
mock_response.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.put.return_value = mock_response
mock_client.__aenter__.return_value = mock_client
mock_client.__aexit__.return_value = None
mock_client_class.return_value = mock_client
await client.redeploy_stack(1, 1, pull_image=True)
call_args = mock_client.put.call_args
assert call_args[1]["json"]["pullImage"] is True
class TestPortainerClientUpdateStackEnv:
"""Test stack environment variable updates."""
@pytest.mark.asyncio
async def test_update_stack_env_sends_env_vars(self):
"""update_stack_env should update environment variables."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
env_vars = [{"name": "DB_HOST", "value": "localhost"}]
with patch.object(client, "get_stack_file", new_callable=AsyncMock) as mock_file:
mock_file.return_value = "version: '3'"
with patch("httpx.AsyncClient") as mock_client_class:
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"Id": 1}
mock_response.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.put.return_value = mock_response
mock_client.__aenter__.return_value = mock_client
mock_client.__aexit__.return_value = None
mock_client_class.return_value = mock_client
await client.update_stack_env(1, 1, env_vars)
call_args = mock_client.put.call_args
assert call_args[1]["json"]["env"] == env_vars
class TestPortainerClientDeleteContainer:
"""Test container deletion."""
@pytest.mark.asyncio
async def test_delete_container_returns_true(self):
"""delete_container should return True on success."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
with patch("httpx.AsyncClient") as mock_client_class:
mock_response = MagicMock()
mock_response.status_code = 204
mock_response.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.delete.return_value = mock_response
mock_client.__aenter__.return_value = mock_client
mock_client.__aexit__.return_value = None
mock_client_class.return_value = mock_client
result = await client.delete_container(1, "abc123")
assert result is True
@pytest.mark.asyncio
async def test_delete_container_with_force(self):
"""delete_container should pass force parameter."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
with patch("httpx.AsyncClient") as mock_client_class:
mock_response = MagicMock()
mock_response.status_code = 204
mock_response.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.delete.return_value = mock_response
mock_client.__aenter__.return_value = mock_client
mock_client.__aexit__.return_value = None
mock_client_class.return_value = mock_client
await client.delete_container(1, "abc123", force=True)
call_args = mock_client.delete.call_args
assert call_args[1]["params"]["force"] == "true"
class TestPortainerClientRestartContainer:
"""Test container restart."""
@pytest.mark.asyncio
async def test_restart_container_returns_true(self):
"""restart_container should return True on success."""
client = PortainerClient(base_url="http://portainer:9000", api_key="key")
with patch("httpx.AsyncClient") as mock_client_class:
mock_response = MagicMock()
mock_response.status_code = 204
mock_response.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.post.return_value = mock_response
mock_client.__aenter__.return_value = mock_client
mock_client.__aexit__.return_value = None
mock_client_class.return_value = mock_client
result = await client.restart_container(1, "abc123")
assert result is True
mock_client.post.assert_called_once()
class TestPortainerClientSingleton: class TestPortainerClientSingleton:
"""Test singleton pattern.""" """Test singleton pattern."""
Executable
+50
View File
@@ -0,0 +1,50 @@
#!/bin/bash
# Core-API Server Startup Script
set -e
# Colors for output
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
RED='\033[0;31m'
NC='\033[0m' # No Color
echo -e "${GREEN}Starting Core-API server...${NC}"
# Check if port 8788 is already in use
if lsof -Pi :8788 -sTCP:LISTEN -t >/dev/null 2>&1 ; then
echo -e "${RED}Error: Port 8788 is already in use${NC}"
echo "Run: lsof -i :8788 to see what's using it"
echo "Or run: kill \$(lsof -t -i:8788) to stop it"
exit 1
fi
# Activate virtual environment if not already activated
if [ -z "$VIRTUAL_ENV" ]; then
if [ -d ".venv" ]; then
echo -e "${YELLOW}Activating virtual environment...${NC}"
source .venv/bin/activate
else
echo -e "${RED}Error: Virtual environment not found${NC}"
echo "Run: python -m venv .venv && source .venv/bin/activate && pip install -r requirements.txt"
exit 1
fi
fi
# Create logs directory if it doesn't exist
LOGS_DIR="logs"
mkdir -p "$LOGS_DIR"
# Clear/create log file
LOG_FILE="$LOGS_DIR/server.log"
> "$LOG_FILE"
echo -e "${YELLOW}Logs will be written to: ${LOG_FILE}${NC}"
# Start the server
echo -e "${GREEN}Starting uvicorn server on http://tower-of-joy:8788${NC}"
echo -e "${YELLOW}Press Ctrl+C to stop the server${NC}"
echo ""
uvicorn src.main:app --reload --host 0.0.0.0 --port 8788 2>&1 | tee "$LOG_FILE"