68 lines
1.6 KiB
Python
68 lines
1.6 KiB
Python
"""
|
|
Base class for vein connectors.
|
|
|
|
Defines the minimal interface all veins should implement.
|
|
The core/ module contains isolated API clients.
|
|
The api/ module wraps them in FastAPI routes.
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Generic, TypeVar
|
|
|
|
from fastapi import APIRouter
|
|
|
|
TCredentials = TypeVar("TCredentials")
|
|
TClient = TypeVar("TClient")
|
|
|
|
|
|
class BaseVein(ABC, Generic[TCredentials, TClient]):
|
|
"""
|
|
Abstract base for vein connectors.
|
|
|
|
Veins are wrappers around API clients that provide:
|
|
- Standard auth patterns (headers or OAuth)
|
|
- Health check endpoint
|
|
- Consistent routing structure
|
|
|
|
The core implementation should be isolated and runnable without FastAPI.
|
|
"""
|
|
|
|
@property
|
|
@abstractmethod
|
|
def name(self) -> str:
|
|
"""Vein name (e.g., 'jira', 'slack', 'google')"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_client(self, creds: TCredentials) -> TClient:
|
|
"""
|
|
Create API client with given credentials.
|
|
|
|
This should delegate to core/ module which contains
|
|
the isolated client implementation.
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def health_check(self, creds: TCredentials) -> dict:
|
|
"""
|
|
Test connection and return status.
|
|
|
|
Should return:
|
|
{
|
|
"status": "ok",
|
|
"user": "...", # or other identifying info
|
|
...
|
|
}
|
|
"""
|
|
pass
|
|
|
|
def create_router(self) -> APIRouter:
|
|
"""
|
|
Create base router with standard endpoints.
|
|
|
|
Subclasses should extend this with additional routes.
|
|
"""
|
|
router = APIRouter()
|
|
return router
|