API Reference for api-service

api_service

FastAPI orchestrator service for the Rune platform.

Modules

dependencies

Dependency injection for FastAPI endpoints.

Functions
get_db 
get_db() -> Generator[Session, None, None]

Provide a database session for dependency injection.

Yields:

Type Description
Session

Database session that automatically commits or rolls back.
Source code in services/api-service/src/api_service/dependencies.py 
10
11
12
13
14
15
16
17
def get_db() -> Generator[Session, None, None]:
    """Provide a database session for dependency injection.

    Yields:
        Database session that automatically commits or rolls back.
    """
    with Session(engine) as session:
        yield session

main

FastAPI application entry point for the API service.

Functions
lifespan async 
lifespan(app: FastAPI)

Manage the lifecycle of the FastAPI application.

Source code in services/api-service/src/api_service/main.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@asynccontextmanager
async def lifespan(app: FastAPI):
    """Manage the lifecycle of the FastAPI application."""
    # Startup
    logger.info("Starting up API service...")
    logger.info("Initializing database...")
    create_db_and_tables()
    logger.info("Database initialized successfully")
    yield
    # Shutdown - wait for background tasks
    if _running_tasks:
        logger.info(f"Waiting for {len(_running_tasks)} tasks to complete...")
        await asyncio.gather(*_running_tasks, return_exceptions=True)
    logger.info("Shutdown complete")
health_check async 
health_check()

Liveness probe - is the service running?

Source code in services/api-service/src/api_service/main.py 
62
63
64
65
@app.get("/health")
async def health_check():
    """Liveness probe - is the service running?"""
    return {"status": "healthy"}
readiness_check async 
readiness_check(db: Session = Depends(get_db))

Readiness probe - can the service handle requests?

Source code in services/api-service/src/api_service/main.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
@app.get("/ready")
async def readiness_check(db: Session = Depends(get_db)):
    """Readiness probe - can the service handle requests?"""
    try:
        # Test database connection
        db.execute(text("SELECT 1"))
        return {"status": "ready", "database": "connected"}
    except Exception as e:
        logger.error(f"Readiness check failed: {e}")
        return JSONResponse(
            status_code=503,
            content={"status": "not_ready", "error": "Database unavailable"},
        )
root async 
root()

Returns a welcome message for the API.

Source code in services/api-service/src/api_service/main.py 
83
84
85
86
@app.get("/")
async def root():
    """Returns a welcome message for the API."""
    return {"message": "Welcome to the API Service"}

routers

API router modules for adapters and sessions.

Modules
adapters

Adapter management router stubs.

Functions
list_adapters async 
list_adapters() -> JSONResponse

List all stored adapters.

Returns:

Type Description
JSONResponse

JSONResponse with list of adapter records.

Raises:

Type Description
HTTPException

501 while not yet implemented.
Example

response = client.get("/adapters") response.status_code 200 isinstance(response.json(), list) True 'adapter_id' in response.json()[0] True

Source code in services/api-service/src/api_service/routers/adapters.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@router.get("")
async def list_adapters() -> JSONResponse:
    """List all stored adapters.

    Returns:
        JSONResponse with list of adapter records.

    Raises:
        HTTPException: 501 while not yet implemented.

    Example:
        >>> response = client.get("/adapters")
        >>> response.status_code
        200
        >>> isinstance(response.json(), list)
        True
        >>> 'adapter_id' in response.json()[0]
        True
    """
    return JSONResponse(status_code=501, content={"detail": "Not Implemented"})
get_adapter async 
get_adapter(adapter_id: str) -> JSONResponse

Get adapter by ID.

Parameters:

Name Type Description Default
adapter_id str

Unique identifier for the adapter.

 required

Returns:

Type Description
JSONResponse

JSONResponse with a single adapter record dict.

Raises:

Type Description
HTTPException

404 if adapter not found.
HTTPException

501 while not yet implemented.
Example

response = client.get("/adapters/test-adapter-123") response.status_code 200 'adapter_id' in response.json() True

Source code in services/api-service/src/api_service/routers/adapters.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@router.get("/{adapter_id}")
async def get_adapter(adapter_id: str) -> JSONResponse:
    """Get adapter by ID.

    Args:
        adapter_id: Unique identifier for the adapter.

    Returns:
        JSONResponse with a single adapter record dict.

    Raises:
        HTTPException: 404 if adapter not found.
        HTTPException: 501 while not yet implemented.

    Example:
        >>> response = client.get("/adapters/test-adapter-123")
        >>> response.status_code
        200
        >>> 'adapter_id' in response.json()
        True
    """
    return JSONResponse(status_code=501, content={"detail": "Not Implemented"})
create_adapter async 
create_adapter() -> JSONResponse

Store a new adapter.

Returns:

Type Description
JSONResponse

JSONResponse with the created adapter record.

Raises:

Type Description
HTTPException

422 if request body is invalid.
HTTPException

501 while not yet implemented.
Example

body = {"adapter_id": "new-adapter-001", "task_type": "bug-fix"} response = client.post("/adapters", json=body) response.status_code 201 'adapter_id' in response.json() True

Source code in services/api-service/src/api_service/routers/adapters.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@router.post("")
async def create_adapter() -> JSONResponse:
    """Store a new adapter.

    Returns:
        JSONResponse with the created adapter record.

    Raises:
        HTTPException: 422 if request body is invalid.
        HTTPException: 501 while not yet implemented.

    Example:
        >>> body = {"adapter_id": "new-adapter-001", "task_type": "bug-fix"}
        >>> response = client.post("/adapters", json=body)
        >>> response.status_code
        201
        >>> 'adapter_id' in response.json()
        True
    """
    return JSONResponse(status_code=501, content={"detail": "Not Implemented"})
sessions

Coding session router stubs.

Functions
list_sessions async 
list_sessions() -> JSONResponse

List coding sessions.

Returns:

Type Description
JSONResponse

JSONResponse with list of coding session records.

Raises:

Type Description
HTTPException

501 while not yet implemented.
Example

response = client.get("/sessions") response.status_code 200 isinstance(response.json(), list) True 'session_id' in response.json()[0] True

Source code in services/api-service/src/api_service/routers/sessions.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@router.get("")
async def list_sessions() -> JSONResponse:
    """List coding sessions.

    Returns:
        JSONResponse with list of coding session records.

    Raises:
        HTTPException: 501 while not yet implemented.

    Example:
        >>> response = client.get("/sessions")
        >>> response.status_code
        200
        >>> isinstance(response.json(), list)
        True
        >>> 'session_id' in response.json()[0]
        True
    """
    return JSONResponse(status_code=501, content={"detail": "Not Implemented"})
get_session async 
get_session(session_id: str) -> JSONResponse

Get coding session by ID.

Parameters:

Name Type Description Default
session_id str

Unique identifier for the coding session.

 required

Returns:

Type Description
JSONResponse

JSONResponse with a single coding session record dict.

Raises:

Type Description
HTTPException

404 if session not found.
HTTPException

501 while not yet implemented.
Example

response = client.get("/sessions/test-session-456") response.status_code 200 'session_id' in response.json() True

Source code in services/api-service/src/api_service/routers/sessions.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@router.get("/{session_id}")
async def get_session(session_id: str) -> JSONResponse:
    """Get coding session by ID.

    Args:
        session_id: Unique identifier for the coding session.

    Returns:
        JSONResponse with a single coding session record dict.

    Raises:
        HTTPException: 404 if session not found.
        HTTPException: 501 while not yet implemented.

    Example:
        >>> response = client.get("/sessions/test-session-456")
        >>> response.status_code
        200
        >>> 'session_id' in response.json()
        True
    """
    return JSONResponse(status_code=501, content={"detail": "Not Implemented"})
create_session async 
create_session() -> JSONResponse

Create a new coding session.

Returns:

Type Description
JSONResponse

JSONResponse with the created coding session record.

Raises:

Type Description
HTTPException

422 if request body is invalid.
HTTPException

501 while not yet implemented.
Example

body = {"session_id": "new-session-001", "task_type": "bug-fix"} response = client.post("/sessions", json=body) response.status_code 201 'session_id' in response.json() True

Source code in services/api-service/src/api_service/routers/sessions.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@router.post("")
async def create_session() -> JSONResponse:
    """Create a new coding session.

    Returns:
        JSONResponse with the created coding session record.

    Raises:
        HTTPException: 422 if request body is invalid.
        HTTPException: 501 while not yet implemented.

    Example:
        >>> body = {"session_id": "new-session-001", "task_type": "bug-fix"}
        >>> response = client.post("/sessions", json=body)
        >>> response.status_code
        201
        >>> 'session_id' in response.json()
        True
    """
    return JSONResponse(status_code=501, content={"detail": "Not Implemented"})

storage

Database storage configuration.

Functions
create_db_and_tables 
create_db_and_tables() -> None

Create all database tables.

Source code in services/api-service/src/api_service/storage.py 
 9
10
11
def create_db_and_tables() -> None:
    """Create all database tables."""
    SQLModel.metadata.create_all(engine)