vs-server
Multi-protocol server framework for Viveka Sutra — HTTP, WebSocket, and SSE on the same class, with pluggable server implementations, lifecycle hooks, and declarative auth guards.
Overview
vs-server is the network layer every VS service is built on. It eliminates framework boilerplate — no manual FastAPI() setup, no router wiring, no middleware calls. Controllers are plain classes annotated with @controller. WebSocket and SSE handlers follow the same pattern with @websocket and @sse. A single class can carry all three protocols at once by stacking decorators.
Server implementations are pluggable via VsServerFactory — register any VsServer subclass under a key and retrieve it at startup. VsFastApiServer ships as the built-in implementation and auto-registers as "fastapi" on import.
The Problem It Solves
Every service needs routing, middleware, exception handling, and lifecycle management. Without vs-server, each service repeats the same boilerplate.
Without vs-server
from fastapi import FastAPI, APIRouter
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app):
await create_tables()
yield
app = FastAPI(title="my-service", lifespan=lifespan)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
router = APIRouter(prefix="/users")
@router.get("")
async def list_users(): ...
@router.post("")
async def create_user(body: UserRequest): ...
app.include_router(router)
uvicorn.run(app, host="0.0.0.0", port=8000)
With vs-server
from vs_server.decorator.vs_controller_decorator import controller, get, post
@controller("/users")
class UserController:
@get("")
async def list_users(self): ...
@post("")
async def create_user(self, body: UserRequest): ...
from vs_server.lifecycle.vs_lifecycle import startup
@startup
async def create_tables():
await VsDbSessionFactory.get_instance().create_tables()
server = VsServerFactory.get("fastapi", config)
server.add_controller(UserController())
server.run()
One pattern, every service, zero drift.
Installation
pip install vs-server
With FastAPI support:
pip install vs-server[fastapi]
Dependencies
| Library | Required | Purpose |
|---|---|---|
vs-common |
Yes | Config, logging |
pydantic |
Yes | Request/response schema validation |
fastapi |
No — install with [fastapi] extra |
FastAPI HTTP server implementation |
uvicorn |
No — install with [fastapi] extra |
ASGI server |
starlette |
No — pulled in by FastAPI | CORS middleware |
vs-db |
No | Database session middleware (register VsDbMiddleware manually) |
vs-security |
No | JWT auth for guards |
Configuration
All server config lives under the [server] section of config.ini.
| Key | Default | Description |
|---|---|---|
server.name |
vs-server |
Server name shown in logs, /, and Swagger |
server.version |
0.1.0 |
Version shown in logs, /, and Swagger |
server.description |
App server developed by Viveka Sutra |
Description shown in Swagger |
server.base_url |
"" |
Base URL prefix for all core routes (/, /health, /vs-docs) |
server.host |
127.0.0.1 |
Host to bind |
server.port |
8000 |
Port to bind |
server.reload |
false |
Enable hot reload (dev only) |
server.workers |
1 |
Number of worker processes |
server.log_level |
info |
Uvicorn log level |
server.ssl_certfile |
— | Path to TLS certificate file |
server.ssl_keyfile |
— | Path to TLS private key file |
server.ssl_keyfile_password |
— | Password for encrypted private key |
api.cors_origins |
* |
Comma-separated allowed CORS origins |
config.ini example:
[server]
name = my-service
version = 1.0.0
host = 0.0.0.0
port = 8000
workers = 2
ssl_certfile = certs/server.crt
ssl_keyfile = certs/server.key
[api]
cors_origins = https://app.example.com,https://admin.example.com
[logging]
level = INFO
file_path = ./logs/service.log
Quick Start
from vs_common.config.vs_ini_config import VsIniConfig
from vs_common.log.vs_log_manager import VsLogManager
from vs_common.schema.vs_log_config import VsLogConfig
from vs_server.server.vs_fast_api_server import VsFastApiServer # noqa — auto-registers "fastapi"
from vs_server.factory.vs_server_factory import VsServerFactory
from vs_server.decorator.vs_server_registry import server_registry
import my_service.controllers # noqa — triggers @controller registration
@server_registry(hooks="my_service.lifecycle")
def main():
config = VsIniConfig("config.ini")
VsLogManager.init(VsLogConfig(level=config.get("logging.level", default="INFO")))
server = VsServerFactory.get("fastapi", config)
server.add_controller(UserController())
server.run()
if __name__ == "__main__":
main()
How It All Fits Together
Application Startup
└── import VsFastApiServer # auto-registers "fastapi" into VsServerFactory
@server_registry(hooks="my_service.lifecycle")
└── imports my_service.lifecycle # @startup / @shutdown self-register into hook lists
VsServerFactory.get("fastapi", config)
├── reads name/version/host/port/TLS from config.ini
├── wires CORS, exception handlers, lifespan
└── exposes /, /health, /vs-docs
server.get_app().add_middleware(VsDbMiddleware) # register manually if using vs-db
server.add_controller(UserController())
└── reads @controller + @get/@post/... → wires FastAPI router
server.run()
├── calls @startup hooks in order
├── starts uvicorn
└── calls @shutdown hooks on stop
HTTP Controllers
Annotate a class with @controller and methods with HTTP verb decorators.
from vs_server.decorator.vs_controller_decorator import controller, get, post, put, delete, patch
from pydantic import BaseModel
class UserRequest(BaseModel):
name: str
email: str
@controller("/users", description="User management endpoints")
class UserController:
@get("", response_model=list[UserResponse])
async def list_users(self): ...
@get("/{user_id}", response_model=UserResponse)
async def get_user(self, user_id: str): ...
@post("", status_code=201, response_model=UserResponse)
async def create_user(self, body: UserRequest): ...
@put("/{user_id}", response_model=UserResponse)
async def update_user(self, user_id: str, body: UserRequest): ...
@delete("/{user_id}", status_code=204)
async def delete_user(self, user_id: str): ...
Register via server.add_controller():
server.add_controller(UserController())
All FastAPI route parameters (response_model, status_code, tags, summary, etc.) pass through via **kwargs.
@controller parameters
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
path |
str |
Yes | — | URL prefix for all routes in this controller |
description |
str |
No | "" |
Tag description shown in Swagger |
guards |
List[Callable] |
No | [] |
Applied to every route in this controller |
HTTP verb decorators
| Decorator | HTTP Method |
|---|---|
@get(path, guards=[], **kwargs) |
GET |
@post(path, guards=[], **kwargs) |
POST |
@put(path, guards=[], **kwargs) |
PUT |
@delete(path, guards=[], **kwargs) |
DELETE |
@patch(path, guards=[], **kwargs) |
PATCH |
Guards on a method are merged with controller-level guards — both run.
WebSocket
Annotate a class with @websocket and message handlers with @on_message.
from vs_server.decorator.vs_websocket_decorator import websocket, on_message
@websocket("/ws/chat", description="Real-time chat over WebSocket")
class ChatHandler:
@on_message(
description="Handle incoming chat message",
args={"text": "Message text", "room": "Target room ID"},
)
async def handle_message(self, data: dict):
text = data.get("text")
# process and send response
@on_message(description="Handle ping")
async def handle_ping(self, data: dict):
...
Register via server.add_websocket():
server.add_websocket(ChatHandler())
Connection lifecycle: Guards run at connect time. If any guard raises, the connection is closed with code 1008 before accept(). Message guards run per-message before the handler.
@websocket parameters
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
path |
str |
Yes | — | WebSocket URL path |
description |
str |
No | "" |
Shown in /vs-docs |
guards |
List[Callable] |
No | [] |
Run at connection time — signature: async guard(websocket) |
@on_message parameters
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
description |
str |
No | "" |
Shown in /vs-docs |
args |
dict |
No | {} |
Argument descriptions shown in /vs-docs |
guards |
List[Callable] |
No | [] |
Run per-message — signature: async guard(websocket, data) |
Server-Sent Events (SSE)
Annotate a class with @sse and the stream method with @sse_stream.
from fastapi import Request
from vs_server.decorator.vs_sse_decorator import sse, sse_stream
@sse("/events/notifications", description="Notification stream")
class NotificationStream:
@sse_stream(
description="Stream live notifications",
args={"user_id": "Filter notifications for this user"},
)
async def stream(self, request: Request):
while not await request.is_disconnected():
event = await notification_queue.get()
yield event
Register via server.add_sse():
server.add_sse(NotificationStream())
The stream method must be an async generator. Each value yielded is sent as a data: <value>\n\n SSE event.
@sse parameters
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
path |
str |
Yes | — | SSE endpoint URL path |
description |
str |
No | "" |
Shown in /vs-docs |
guards |
List[Callable] |
No | [] |
Run before the stream starts |
@sse_stream parameters
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
description |
str |
No | "" |
Shown in /vs-docs |
args |
dict |
No | {} |
Argument descriptions shown in /vs-docs |
guards |
List[Callable] |
No | [] |
Run before the stream starts |
Multi-Protocol Classes
A single class can carry HTTP, WebSocket, and SSE by stacking decorators:
from vs_server.decorator.vs_controller_decorator import controller, get
from vs_server.decorator.vs_websocket_decorator import websocket, on_message
from vs_server.decorator.vs_sse_decorator import sse, sse_stream
@controller("/chat", description="Chat HTTP endpoints")
@websocket("/ws/chat", description="Chat WebSocket")
@sse("/events/chat", description="Chat event stream")
class ChatHandler:
@get("/history")
async def get_history(self): ...
@on_message(description="Send a message")
async def handle_message(self, data: dict): ...
@sse_stream(description="Live chat events")
async def stream(self, request): ...
Register each protocol separately:
handler = ChatHandler()
server.add_controller(handler)
server.add_websocket(handler)
server.add_sse(handler)
Guards
Guards are async callables that run before a route, WebSocket connection, or SSE stream. If a guard raises, the request is rejected.
from vs_security.guard.vs_security_factory import VsSecurityFactory
# controller-level guard — applies to all routes
@controller("/admin", guards=[VsSecurityFactory.get()])
class AdminController:
# route-level guard — merged with controller guards
@get("/report", guards=[require_admin_role])
async def get_report(self): ...
Guard signature by protocol:
| Protocol | Point | Guard signature |
|---|---|---|
| HTTP | Per-request | async def guard() -> None (FastAPI Depends) |
| WebSocket | At connect | async def guard(websocket: WebSocket) -> None |
| WebSocket | Per-message | async def guard(websocket: WebSocket, data: dict) -> None |
| SSE | Before stream | async def guard() -> None (FastAPI Depends) |
Guards that raise PermissionError are handled by the built-in exception handler and return a 403 response on HTTP. On WebSocket, the connection is closed with code 1008.
Lifecycle Hooks
Use @startup and @shutdown to run code when the server starts and stops.
from vs_server.lifecycle.vs_lifecycle import startup, shutdown
from vs_db.session.vs_db_session_factory import VsDbSessionFactory
@startup
async def create_tables():
await VsDbSessionFactory.get_instance().create_tables()
@shutdown
async def close_connections():
await VsDbSessionFactory.get_instance().close()
Hooks self-register into global lists at import time. They run in registration order. Both sync and async hooks are supported.
Register hook modules with server_registry so they are imported before the server starts:
from vs_server.decorator.vs_server_registry import server_registry
@server_registry(hooks="my_service.lifecycle")
def main():
...
VsServer.startup() iterates _startup_hooks. VsServer.shutdown() iterates _shutdown_hooks.
VsServerFactory and @server
VsServerFactory maps string keys to VsServer subclasses. Register once, retrieve by key anywhere.
from vs_server.factory.vs_server_factory import VsServerFactory
from vs_server.server.vs_fast_api_server import VsFastApiServer # noqa — auto-registers "fastapi"
server = VsServerFactory.get("fastapi", config)
Adding a custom server implementation
from vs_server.server.vs_server import VsServer
from vs_server.decorator.vs_server_decorator import server
from vs_server.factory.vs_server_factory import VsServerFactory
@server("flask")
class VsFlaskServer(VsServer):
def add_controller(self, instance, guards=None): ...
def add_router(self, router, prefix=""): ...
def add_websocket(self, instance): ...
def add_sse(self, instance): ...
def get_app(self): ...
def run(self): ...
VsServerFactory.register("flask", VsFlaskServer)
Or scan the package automatically with server_registry:
@server_registry(server="my_pkg.server")
def main():
server = VsServerFactory.get("flask", config)
server.run()
server_registry
Scans packages for @server-decorated classes and @startup/@shutdown hook functions. Works as a decorator on main() or as a plain call.
from vs_server.decorator.vs_server_registry import server_registry
# as a decorator
@server_registry(server="my_pkg.server", hooks="my_pkg.lifecycle")
def main():
...
# as a plain call
server_registry(hooks="my_service.lifecycle")
| Parameter | Description |
|---|---|
server |
Package path to scan for @server-decorated classes. Not needed for "fastapi" — it auto-registers on import. |
hooks |
Package path to import for @startup/@shutdown registration. |
HTTPS / TLS
Set ssl_certfile and ssl_keyfile in config.ini. No code changes required.
[server]
ssl_certfile = certs/server.crt
ssl_keyfile = certs/server.key
ssl_keyfile_password = optional-passphrase
TLS is enabled when both ssl_certfile and ssl_keyfile are present. The startup banner prints TLS: enabled.
Built-in Endpoints
Every VsFastApiServer instance exposes three endpoints automatically:
| Endpoint | Description |
|---|---|
GET / |
Returns {"service": name, "version": version, "status": "running"} |
GET /health |
Returns VsServerHealth with overall status, uptime, cache, and dependency health |
GET /vs-docs |
Custom HTML docs page for WebSocket and SSE endpoints |
If server.base_url is set in config.ini, all three are prefixed with it.
/health response
{
"status": "healthy",
"name": "my-service",
"version": "1.0.0",
"uptime_seconds": 142.3,
"cache": "healthy",
"instance_id": "hostname",
"dependencies": {
"database": "healthy"
},
"timestamp": "2026-01-01T00:00:00Z"
}
Overall status logic:
| Condition | Status |
|---|---|
| All dependencies healthy | healthy |
| Some healthy, some unhealthy | degraded |
| All unhealthy | unhealthy |
Override _check_dependencies() in a VsServer subclass to include custom dependency checks:
async def _check_dependencies(self) -> dict:
try:
await db.execute("SELECT 1")
return {"database": "healthy"}
except Exception:
return {"database": "unhealthy"}
Documentation
| URL | Content |
|---|---|
/docs |
Swagger UI — HTTP endpoints only (FastAPI native) |
/vs-docs |
Custom HTML page — WebSocket and SSE endpoints |
HTTP endpoints are documented automatically via FastAPI's Swagger UI. WebSocket and SSE endpoints are not supported by OpenAPI — they appear in /vs-docs instead, generated by VsDocsRenderer.
get_docs_url() returns the framework-native docs URL ("/docs" for FastAPI). Override to change or suppress it.
Using with vs-db
Initialize VsDbSessionFactory before creating the server, then register VsDbMiddleware explicitly:
from vs_db.session.vs_db_session_factory import VsDbSessionFactory
from vs_db.middleware.vs_db_middleware import VsDbMiddleware
from vs_server.server.vs_fast_api_server import VsFastApiServer # noqa — auto-registers "fastapi"
from vs_server.factory.vs_server_factory import VsServerFactory
def main():
config = VsIniConfig("config.ini")
VsDbSessionFactory(config)
server = VsServerFactory.get("fastapi", config)
server.get_app().add_middleware(VsDbMiddleware)
server.add_controller(UserController())
server.run()
VsDbMiddleware must be registered before server.run() — middleware cannot be added after the server starts accepting requests.
Error Handling
Raise exceptions from vs_server.schema.exceptions inside controllers — the built-in handlers convert them to structured JSON responses automatically.
from vs_server.schema.exceptions import NotFoundException, ValidationException
@controller("/users")
class UserController:
@get("/{user_id}")
async def get_user(self, user_id: str):
user = await repo.find(user_id)
if user is None:
raise NotFoundException(f"User '{user_id}' not found")
return user
Error response format:
{
"success": false,
"error": {
"code": "NOT_FOUND",
"message": "User 'abc' not found",
"details": {}
}
}
Built-in exception types:
| Exception | HTTP Status | Error Code | Default Message |
|---|---|---|---|
NotFoundException |
404 | NOT_FOUND |
Resource not found |
ValidationException |
400 | VALIDATION_ERROR |
Invalid request |
UnauthorizedException |
401 | UNAUTHORIZED |
Authentication required |
ForbiddenException |
403 | FORBIDDEN |
Access denied |
InternalServerException |
500 | INTERNAL_ERROR |
Internal server error |
ServiceUnavailableException |
503 | SERVICE_UNAVAILABLE |
Service unavailable |
RateLimitException |
429 | RATE_LIMIT_EXCEEDED |
Rate limit exceeded |
TimeoutException |
504 | TIMEOUT_ERROR |
Request timed out |
All extend VsAPIException. Catch by base type to handle errors uniformly.
Automatically handled Python exceptions:
| Exception | HTTP Status |
|---|---|
PermissionError |
403 |
ValueError |
400 |
fastapi.RequestValidationError |
400 (with field-level detail) |
Any other Exception |
500 |
Extending vs-server
Implement VsServer to add support for any framework. The abstract interface is:
from vs_server.server.vs_server import VsServer
from vs_server.config.vs_base_config import VsBaseConfig
class MyServer(VsServer):
def __init__(self, config: VsBaseConfig):
super().__init__(config)
def add_controller(self, instance, guards=None) -> None: ...
def add_router(self, router, prefix: str = "") -> None: ...
def add_websocket(self, instance) -> None: ...
def add_sse(self, instance) -> None: ...
def get_app(self): ...
def run(self) -> None: ...
Override template methods to customise behaviour:
| Method | Default | Override to |
|---|---|---|
startup() |
Logs + runs @startup hooks |
Add custom startup logic. Call super().startup() to keep hooks. |
shutdown() |
Runs @shutdown hooks + logs |
Add custom shutdown logic. Call super().shutdown() to keep hooks. |
get_docs_url() |
Returns None |
Return the framework-native docs URL |
_check_dependencies() |
Returns {} |
Return dict[str, "healthy" | "unhealthy"] merged into /health |
_register_middleware(app) |
Adds CORS | Add custom middleware. Call super() to keep CORS. |
_register_exception_handlers(app) |
Adds built-in handlers | Add custom exception handlers. Call super() to keep built-ins. |
Class Reference
VsServer
Abstract base class for all server implementations. Manages config, logging, and lifecycle hooks.
Constructor:
| Parameter | Type | Description |
|---|---|---|
config |
VsBaseConfig |
Application config — reads server.* keys |
Attributes set from config:
| Attribute | Config key | Default |
|---|---|---|
name |
server.name |
vs-server |
version |
server.version |
0.1.0 |
description |
server.description |
App server developed by Viveka Sutra |
base_url |
server.base_url |
"" |
Methods:
| Method | Signature | Description |
|---|---|---|
startup |
async startup() -> None |
Logs startup banner, then runs all @startup hooks. Call super().startup() to keep. |
shutdown |
async shutdown() -> None |
Runs all @shutdown hooks, then logs stop. Call super().shutdown() to keep. |
get_docs_url |
get_docs_url() -> Optional[str] |
Returns None. Override to return framework-native docs URL. |
add_controller |
add_controller(instance, guards=None) -> None |
Abstract. Register a @controller instance. |
add_router |
add_router(router, prefix="") -> None |
Abstract. Register a raw framework router. |
add_websocket |
add_websocket(instance) -> None |
Abstract. Register a @websocket instance. |
add_sse |
add_sse(instance) -> None |
Abstract. Register an @sse instance. |
get_app |
get_app() -> Any |
Abstract. Return the underlying framework app. |
run |
run() -> None |
Abstract. Start the server. |
VsServerFactory
Registry mapping string keys to VsServer subclasses. Thread-safe class-level dict.
Methods:
| Method | Signature | Description |
|---|---|---|
register |
register(key: str, server_class: Type[VsServer]) -> None |
Register a server class. Raises TypeError if not a VsServer subclass. |
get |
get(key: str, config: VsBaseConfig) -> VsServer |
Instantiate and return a server. Raises KeyError if key not registered. |
Notes:
VsFastApiServerauto-registers as"fastapi"when its module is imported.- Import
VsFastApiServerbefore callingVsServerFactory.get("fastapi", config).
VsFastApiServer
Built-in VsServer implementation backed by FastAPI and uvicorn. Auto-registers as "fastapi" on import.
Behaviour at construction:
- Raises
ImportErroriffastapiis not installed. - Builds the FastAPI app with lifespan, CORS middleware, exception handlers, and core routes.
Built-in routes: GET /, GET /health, GET /vs-docs.
Methods (beyond VsServer contract):
| Method | Description |
|---|---|
get_docs_url() |
Returns "/docs" |
_check_dependencies() |
Override to add custom dependency health checks |
_register_middleware(app) |
Adds CORS. Call super() to keep it. |
_register_exception_handlers(app) |
Adds built-in handlers. Call super() to keep them. |
Notes:
reload=Trueandworkers > 1are mutually exclusive. Ifreloadistrue,workersis ignored.- TLS is enabled when both
ssl_certfileandssl_keyfileare set. VsDbMiddlewaremust be registered manually:server.get_app().add_middleware(VsDbMiddleware).
@server
Decorator. Marks a class as a named server implementation. Validates that it extends VsServer.
Parameters:
| Parameter | Type | Required | Description |
|---|---|---|---|
key |
str |
Yes | Registry key used with VsServerFactory.get(key, config) |
server_registry
Scans packages for @server-decorated classes and @startup/@shutdown hook functions. Works as a function decorator or a plain call. All parameters are optional.
Parameters:
| Parameter | Type | Description |
|---|---|---|
server |
Optional[str] |
Package path to scan for @server classes |
hooks |
Optional[str] |
Package path to import for lifecycle hook registration |
@startup / @shutdown
Decorators. Register a function into _startup_hooks or _shutdown_hooks at import time. Both sync and async functions are supported.
from vs_server.lifecycle.vs_lifecycle import startup, shutdown
@startup
async def on_start(): ...
@shutdown
def on_stop(): ...
Hooks run in registration order via VsServer.startup() and VsServer.shutdown().
@controller
Decorator. Marks a class as an HTTP controller. Sets cls._vs_controller with path, description, and guards.
Parameters:
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
path |
str |
Yes | — | URL prefix for all routes in this controller |
description |
str |
No | "" |
Tag description shown in Swagger |
guards |
List[Callable] |
No | [] |
Applied to every route in this controller |
@get, @post, @put, @delete, @patch
Route decorators. Applied to methods inside a @controller class.
Parameters (all share the same signature):
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
path |
str |
No | "" |
Route path appended to the controller prefix |
guards |
List[Callable] |
No | [] |
Merged with controller guards |
**kwargs |
— | No | — | Passed to the underlying FastAPI route decorator (e.g. response_model, status_code, summary) |
@websocket
Decorator. Marks a class as a WebSocket handler.
Parameters:
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
path |
str |
Yes | — | WebSocket URL path |
description |
str |
No | "" |
Shown in /vs-docs |
guards |
List[Callable] |
No | [] |
Run at connect time — async guard(websocket) |
@on_message
Decorator. Marks a method inside a @websocket class as a message handler.
Parameters:
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
description |
str |
No | "" |
Shown in /vs-docs |
args |
dict |
No | {} |
{arg_name: description} shown in /vs-docs |
guards |
List[Callable] |
No | [] |
Run per-message — async guard(websocket, data) |
@sse
Decorator. Marks a class as an SSE handler.
Parameters:
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
path |
str |
Yes | — | SSE endpoint URL path |
description |
str |
No | "" |
Shown in /vs-docs |
guards |
List[Callable] |
No | [] |
Run before the stream starts |
@sse_stream
Decorator. Marks a method inside an @sse class as the event stream generator.
Parameters:
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
description |
str |
No | "" |
Shown in /vs-docs |
args |
dict |
No | {} |
{arg_name: description} shown in /vs-docs |
guards |
List[Callable] |
No | [] |
Run before the stream starts |
VsAPIException
Base exception class for all HTTP error responses. All subclasses are handled automatically by VsFastApiServer.
Constructor:
| Parameter | Type | Description |
|---|---|---|
status_code |
int |
HTTP status code |
error_code |
str |
Machine-readable error code |
message |
str |
Human-readable error message |
details |
Optional[Dict] |
Additional error context |
Subclasses:
| Class | Status | Code |
|---|---|---|
NotFoundException |
404 | NOT_FOUND |
ValidationException |
400 | VALIDATION_ERROR |
UnauthorizedException |
401 | UNAUTHORIZED |
ForbiddenException |
403 | FORBIDDEN |
InternalServerException |
500 | INTERNAL_ERROR |
ServiceUnavailableException |
503 | SERVICE_UNAVAILABLE |
RateLimitException |
429 | RATE_LIMIT_EXCEEDED |
TimeoutException |
504 | TIMEOUT_ERROR |
RateLimitException accepts an additional retry_after: int parameter (seconds) and sets the Retry-After response header.
VsServerHealth
Pydantic model returned by GET /health.
Fields:
| Field | Type | Description |
|---|---|---|
status |
str |
healthy, degraded, or unhealthy |
name |
str |
Server name |
version |
str |
Server version |
uptime_seconds |
Optional[float] |
Seconds since startup |
cache |
Optional[str] |
healthy or unhealthy |
instance_id |
Optional[str] |
Hostname of the running instance |
dependencies |
Dict[str, str] |
Custom dependency statuses from _check_dependencies() |
timestamp |
datetime |
UTC timestamp of the health check |
VsDocsRenderer
Generates the /vs-docs HTML page for WebSocket and SSE endpoints.
Constructor:
| Parameter | Type | Description |
|---|---|---|
name |
str |
Server name shown in the page header |
version |
str |
Server version shown in the page header |
Methods:
| Method | Signature | Description |
|---|---|---|
render |
render(ws_endpoints: list, sse_endpoints: list) -> str |
Returns a complete HTML string for the docs page |
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file vs_server-0.1.1.tar.gz.
File metadata
- Download URL: vs_server-0.1.1.tar.gz
- Upload date:
- Size: 36.3 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/7.0.0 CPython/3.11.15
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
fa7f3c14aaedc5ec72db2f314b9d0e8743b594be763f75024fadbb15d373c364
|
|
| MD5 |
093152be14760c9a66c018dcd556834d
|
|
| BLAKE2b-256 |
1fe2d798aa94d16afb3c41ffcd2f421ef55fd24eec6dba264ad3c1ff587a3167
|
File details
Details for the file vs_server-0.1.1-py3-none-any.whl.
File metadata
- Download URL: vs_server-0.1.1-py3-none-any.whl
- Upload date:
- Size: 29.2 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/7.0.0 CPython/3.11.15
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
c9ceeb8f9a9ca6e7aac0531766c198d4af94f97199ee0fd2a0227dee534beb84
|
|
| MD5 |
b0e4cc8c43c2d2c4b5440f6e2bc1c338
|
|
| BLAKE2b-256 |
7727811da3dbbfab1f3a27642e565bf3f291b1d9916edb1697f4ca303aa3dd46
|