File size: 2,327 Bytes
8c290e6 bceb9b7 8c290e6 bceb9b7 8c290e6 bceb9b7 8c290e6 |
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 |
from apscheduler.executors.asyncio import AsyncIOExecutor
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from contextlib import asynccontextmanager
from datetime import datetime
from fastapi import FastAPI, Request
from fastapi.responses import Response, FileResponse
from typing import Any
import httpx
import json
import subprocess
class PrettyJSONResponse(Response):
media_type = "application/json"
def render(self, content: Any) -> bytes:
return json.dumps(content, indent=2).encode("utf-8")
@asynccontextmanager
async def lifespan(app: FastAPI):
scheduler.start()
yield
scheduler.shutdown()
app = FastAPI(lifespan=lifespan)
@app.middleware("http")
async def log_request(request: Request, call_next: Any):
ts = datetime.now().strftime("%y%m%d%H%M%S%f")
data = {
"day": int(ts[:6]),
"dt": int(ts[:-3]),
"url": request.url,
"query_params": request.query_params,
"client": request.client.host,
"method": request.method,
"headers": dict(request.headers),
}
output = json.dumps(
obj=data,
default=str,
indent=None,
separators=(", ", ":"),
)
with open("a.json", "a") as f:
separator = "\n" if f.tell() else ""
f.write(separator + output)
response = await call_next(request)
return response
@app.get("/")
def read_root(request: Request):
return {"Hello": request.client.host}
@app.get("/a", response_class=PrettyJSONResponse)
def get_analytics(n: int = 5):
if n == 0:
cmd = "cat a.json"
else:
cmd = f"tail -{n} a.json"
json_lines = subprocess.run(cmd.split(), capture_output=True).stdout
content = json.loads(f"[{json_lines.replace(b"\n", b",").decode()}]")
return content
@app.api_route("/qa", response_class=FileResponse, methods=["GET", "HEAD"])
def query_analytics():
return "a.json"
@app.get("/favicon.ico")
async def favicon():
return {"message": "woof!"}
@app.get("/ping")
async def ping():
return {"message": "woof!"}
async def self_ping():
async with httpx.AsyncClient() as client:
_ = await client.get("http://0.0.0.0:7860/ping")
scheduler = AsyncIOScheduler(executors={"default": AsyncIOExecutor()})
scheduler.add_job(self_ping, "interval", minutes=60)
|