81 lines
2.2 KiB
Python
81 lines
2.2 KiB
Python
import datetime
|
|
import importlib
|
|
import json
|
|
import logging
|
|
import typing
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
from starlette import status
|
|
import sentry_sdk
|
|
from sentry_sdk.integrations.asgi import SentryAsgiMiddleware
|
|
import seqlog
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
class JsonDateEncoder(json.JSONEncoder):
|
|
def default(self, o: Any) -> Any:
|
|
if isinstance(o, datetime.datetime):
|
|
return o.isoformat()
|
|
return super().default(o)
|
|
|
|
|
|
seqlog.log_to_seq(
|
|
server_url="http://log.abdus.dev:5341/",
|
|
api_key="ePewcJUxe8dST5EiC8Pw",
|
|
level=logging.INFO,
|
|
batch_size=10,
|
|
auto_flush_timeout=10, # seconds
|
|
override_root_logger=True,
|
|
json_encoder_class=JsonDateEncoder, # Optional; only specify this if you want to use a custom JSON encoder
|
|
)
|
|
sentry_sdk.init(
|
|
"https://ccb81025166c449d87e0646cede48448@o271257.ingest.sentry.io/6040020",
|
|
# Set traces_sample_rate to 1.0 to capture 100%
|
|
# of transactions for performance monitoring.
|
|
# We recommend adjusting this value in production.
|
|
traces_sample_rate=1.0,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class HookResponse(BaseModel):
|
|
id: str
|
|
result: typing.Optional[dict] = None
|
|
error: typing.Optional[str] = None
|
|
|
|
|
|
@app.post("/{event}", response_model=HookResponse)
|
|
def receive_hook(event: str, payload: dict):
|
|
logger.info(f"Received hook request for {event=} with {payload=}")
|
|
hook_id = uuid.uuid4().hex
|
|
|
|
try:
|
|
module = importlib.import_module(f"scripts.{event}")
|
|
func = getattr(module, "main")
|
|
except ModuleNotFoundError:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No such handler")
|
|
|
|
if not callable(func):
|
|
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Invalid script interface")
|
|
|
|
try:
|
|
logger.info(f"Running hook handler {func=}")
|
|
result = func(payload)
|
|
return HookResponse(id=hook_id, result=result, error=None)
|
|
except Exception as e:
|
|
logger.error(f"Hook handler threw error", exc_info=True)
|
|
return HookResponse(id=hook_id, result=None, error=str(e))
|
|
|
|
|
|
app = SentryAsgiMiddleware(app)
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
|
|
uvicorn.run(app)
|