Architecture

Architecture

vcpkg-harbor is built with a modular, plugin-based architecture.

Project Structure

src/vcpkg_harbor/
├── __init__.py           # Package version
├── __main__.py           # CLI entry point
├── app.py                # FastAPI application factory
├── core/                 # Core infrastructure
│   ├── config.py         # Configuration management
│   ├── logging.py        # Structured logging
│   ├── exceptions.py     # Custom exceptions
│   └── dependencies.py   # FastAPI dependencies
├── api/                  # REST API endpoints
│   ├── cache.py          # vcpkg protocol (HEAD/GET/PUT)
│   ├── health.py         # Health check endpoints
│   └── metrics.py        # Prometheus metrics
├── storage/              # Storage layer
│   ├── base.py           # StorageBackend protocol
│   ├── registry.py       # Backend discovery
│   └── backends/         # Backend implementations
├── services/             # Business logic
│   ├── cache_service.py  # Package operations
│   ├── stats_service.py  # Statistics
│   └── package_service.py # Package queries
├── auth/                 # Authentication
│   ├── middleware.py     # Auth middleware
│   └── providers.py      # Auth providers
└── dashboard/            # Web UI
    ├── router.py         # Dashboard routes
    └── templates/        # Jinja2 templates

Core Components

Application Factory

app.py contains the create_app() factory that:

1. Loads configuration
2. Sets up logging
3. Initializes storage backend
4. Creates services
5. Configures middleware
6. Registers routes

Configuration

Configuration uses Pydantic Settings with nested models:

class Settings(BaseSettings):
    server: ServerSettings
    storage: StorageSettings
    minio: MinioSettings
    # ...

Environment variables use prefixes: VCPKG_SERVER_PORT, VCPKG_MINIO_ENDPOINT, etc.

Storage Protocol

The StorageBackend protocol defines the interface:

@runtime_checkable
class StorageBackend(Protocol):
    async def initialize(self) -> None: ...
    async def close(self) -> None: ...
    async def exists(self, name, version, sha, triplet, scope=None) -> bool: ...
    async def get(self, name, version, sha, triplet, scope=None) -> AsyncIterator[bytes]: ...
    async def put(self, name, version, sha, triplet, data, size, scope=None) -> PackageInfo: ...
    async def delete(self, name, version, sha, triplet, scope=None) -> bool: ...
    async def stat(self, name, version, sha, triplet, scope=None) -> PackageInfo: ...
    async def list_packages(...) -> list[PackageInfo]: ...
    async def put_metadata(self, key: str, data: bytes) -> None: ...
    async def get_metadata(self, key: str) -> bytes | None: ...
    async def delete_metadata(self, key: str) -> bool: ...
    async def list_metadata(self, prefix: str) -> list[str]: ...
    async def get_stats() -> dict: ...
    async def health_check() -> bool: ...

scope is an optional key prefix; None means the shared identity key {name}/{version}/{sha}/{triplet}, which is also the historical layout. The *_metadata methods store harbor’s own small bookkeeping documents (the build tag index and reference counters) under the reserved _harbor/ prefix, so build tags work identically on every backend. vcpkg_harbor.storage.layout holds the key builders and the parser all backends share.

Backend Discovery

Backends are discovered via entry points:

# pyproject.toml
[project.entry-points."vcpkg_harbor.storage"]
minio = "vcpkg_harbor.storage.backends.minio:MinioBackend"
filesystem = "vcpkg_harbor.storage.backends.filesystem:FilesystemBackend"

Service Layer

Services encapsulate business logic:

• CacheService: Package CRUD with logging
• StatsService: Metrics collection
• PackageService: Package queries for dashboard

Request Flow

graph TB
    A[Client Request] --> B[Auth Middleware]
    B --> C[FastAPI Router]
    C --> D[Service Layer]
    D --> E[Storage Backend]

    style A fill:#4a90e2,stroke:#2c5aa0,color:#fff
    style B fill:#ffa726,stroke:#f57c00,color:#fff
    style C fill:#7b68ee,stroke:#5a4fcf,color:#fff
    style D fill:#50c878,stroke:#3a9d5f,color:#fff
    style E fill:#26c6da,stroke:#00acc1,color:#fff

Async Architecture

All storage operations are async, using asyncio.run_in_executor() for synchronous SDK calls (MinIO, boto3, etc.).

Extensibility

Adding a Backend

1. Implement StorageBackend protocol
2. Add configuration model
3. Register via entry points

Adding Authentication

Implement AuthProvider protocol:

class AuthProvider(ABC):
    async def authenticate(self, request: Request) -> bool: ...
    def get_user(self, request: Request) -> str | None: ...