-
Notifications
You must be signed in to change notification settings - Fork 821
Expand file tree
/
Copy pathmain.py
More file actions
273 lines (221 loc) · 8.33 KB
/
Copy pathmain.py
File metadata and controls
273 lines (221 loc) · 8.33 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
import logging
import re
import time
import uuid
from collections.abc import Awaitable, Callable
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING
import sentry_sdk
from fastapi import FastAPI, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi_pagination import add_pagination
from pydantic import ValidationError
from sentry_sdk.integrations.fastapi import FastApiIntegration
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration
from sentry_sdk.integrations.starlette import StarletteIntegration
from src._version import HONCHO_VERSION
from src.cache.client import close_cache, init_cache
from src.config import settings
from src.db import engine, register_db_query_instrumentation, request_context
from src.exceptions import HonchoException
from src.routers import (
conclusions,
keys,
messages,
peers,
sessions,
webhooks,
workspaces,
)
from src.startup import validate_embedding_schema
from src.telemetry import (
initialize_telemetry_async,
metrics_endpoint,
prometheus_metrics,
register_db_pool_collector,
shutdown_telemetry,
)
from src.telemetry.logging import get_route_template
from src.telemetry.sentry import initialize_sentry
if TYPE_CHECKING:
from sentry_sdk._types import Event, Hint
def get_log_level() -> int:
"""
Convert log level string from settings to logging module constant.
Returns:
int: The logging level constant (e.g., logging.INFO)
"""
log_level_str = settings.LOG_LEVEL.upper()
log_levels = {
"CRITICAL": logging.CRITICAL, # 50
"ERROR": logging.ERROR, # 40
"WARNING": logging.WARNING, # 30
"INFO": logging.INFO, # 20
"DEBUG": logging.DEBUG, # 10
"NOTSET": logging.NOTSET, # 0
}
return log_levels.get(log_level_str, logging.INFO)
# Configure logging
logging.basicConfig(
level=get_log_level(),
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
# Suppress cashews Redis error logs (NoScriptError, ConnectionError, etc.)
# These are handled gracefully by SafeRedis and don't need full tracebacks
logging.getLogger("cashews.backends.redis.client").setLevel(logging.CRITICAL)
class MetricsAccessFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
msg = record.getMessage()
return "GET /metrics" not in msg
logging.getLogger("uvicorn.access").addFilter(MetricsAccessFilter())
def before_send(event: "Event", hint: "Hint | None") -> "Event | None":
"""Filter out events raised from known non-actionable exceptions before Sentry sees them."""
if not hint:
return event
exc_info = hint.get("exc_info")
if not exc_info:
return event
_, exc_value, _ = exc_info
if isinstance(exc_value, HonchoException):
return None
# Filters out ValidationErrors and RequestValidationErrors (typically coming from Pydantic)
if isinstance(exc_value, ValidationError | RequestValidationError):
logger.info(f"Filtering out validation error from Sentry: {exc_value}")
return None
return event
# Sentry Setup
SENTRY_ENABLED = settings.SENTRY.ENABLED
if SENTRY_ENABLED:
initialize_sentry(
integrations=[
StarletteIntegration(
transaction_style="endpoint",
),
FastApiIntegration(
transaction_style="endpoint",
),
# Explicit so DB-query spans are not reliant on auto-enabling.
SqlalchemyIntegration(),
],
before_send=before_send,
)
@asynccontextmanager
async def lifespan(_: FastAPI):
# Initialize CloudEvents telemetry
await initialize_telemetry_async()
# Expose DB connection-pool stats for this API instance (no-op if metrics off)
register_db_pool_collector("api")
register_db_query_instrumentation("api")
# Pre-materialize bounded-label counter children at 0 so metrics are visible
# in Prometheus before the first event (no-op if metrics off).
prometheus_metrics.initialize_bounded_metrics(instance_type="api")
# Validate embedding schema before serving any traffic. Fails closed: if
# the configured EMBEDDING_VECTOR_DIMENSIONS does not match the physical
# pgvector columns, the process refuses to start rather than silently
# writing wrong-dim vectors.
await validate_embedding_schema(engine)
try:
await init_cache()
except Exception as e:
logger.warning(
"Error initializing cache in api process; proceeding without cache: %s", e
)
try:
yield
finally:
# Import here to avoid circular import at module load time
from src.vector_store import close_external_vector_store
await close_external_vector_store()
await close_cache()
await engine.dispose()
# Shutdown telemetry (flush CloudEvents buffer)
await shutdown_telemetry()
app = FastAPI(
lifespan=lifespan,
servers=[
{"url": "https://api.honcho.dev", "description": "Production SaaS Platform"},
{"url": "http://localhost:8000", "description": "Local Development Server"},
],
title="Honcho API",
summary="The Identity Layer for the Agentic World",
description="""Honcho is a platform for giving agents user-centric memory and social cognition.""",
version=HONCHO_VERSION,
contact={
"name": "Plastic Labs",
"url": "https://honcho.dev",
"email": "hello@plasticlabs.ai",
},
license_info={
"name": "GNU Affero General Public License v3.0",
"identifier": "AGPL-3.0-only",
"url": "https://github.com/plastic-labs/honcho/blob/main/LICENSE",
},
)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.CORS_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
add_pagination(app)
app.include_router(workspaces.router, prefix="/v3")
app.include_router(peers.router, prefix="/v3")
app.include_router(sessions.router, prefix="/v3")
app.include_router(messages.router, prefix="/v3")
app.include_router(conclusions.router, prefix="/v3")
app.include_router(keys.router, prefix="/v3")
app.include_router(webhooks.router, prefix="/v3")
# Prometheus metrics endpoint
app.add_route("/metrics", metrics_endpoint, methods=["GET"])
@app.get("/health")
async def health_check():
"""Health check endpoint for monitoring and container orchestration."""
return {"status": "ok"}
# Global exception handlers
@app.exception_handler(HonchoException)
async def honcho_exception_handler(_request: Request, exc: HonchoException):
"""Handle all Honcho-specific exceptions."""
logger.error(f"{exc.__class__.__name__}: {exc.detail}", exc_info=exc)
return JSONResponse(
status_code=exc.status_code,
content={"detail": exc.detail},
)
@app.exception_handler(Exception)
async def global_exception_handler(_request: Request, exc: Exception):
"""Handle all unhandled exceptions."""
logger.error(f"Unhandled exception: {str(exc)}", exc_info=True)
if SENTRY_ENABLED:
sentry_sdk.capture_exception(exc)
return JSONResponse(
status_code=500,
content={"detail": "An unexpected error occurred"},
)
@app.middleware("http")
async def track_request(
request: Request, call_next: Callable[[Request], Awaitable[Response]]
):
# Create a request ID that includes endpoint information
endpoint = re.sub(r"/[A-Za-z0-9_-]{21}", "", request.url.path).replace("/", "_")
request_id = f"{request.method}:{endpoint}:{str(uuid.uuid4())[:8]}"
# Store in request state and context var
request.state.request_id = request_id
token = request_context.set(f"api:{request_id}")
try:
start_time = time.perf_counter()
response = await call_next(request)
# Track metrics if enabled
if settings.METRICS.ENABLED:
template = get_route_template(request)
prometheus_metrics.record_api_request(
method=request.method,
endpoint=template,
status_code=str(response.status_code),
duration_seconds=time.perf_counter() - start_time,
)
return response
finally:
request_context.reset(token)