waylon/fastapi-dynamic-response
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

add structlog

Browse source
This commit is contained in:
Waylon S. Walker 2024-10-16 20:46:11 -05:00
parent 7f0934ac14
commit f64e488ab1
7 changed files with 323 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

114
src/fastapi_dynamic_response/logging_config.py Normal file
View file

@ -0,0 +1,114 @@
# logging_config.py
import logging
import structlog
import sys
from structlog.dev import ConsoleRenderer
from structlog.processors import JSONRenderer
from rich.traceback import install
logger = structlog.get_logger()
install(show_locals=True)
def configure_logging_one(dev_mode: bool = True):
"""Configure structlog based on the mode (dev or prod)."""
logging.basicConfig(
format="%(message)s",
stream=sys.stdout,
level=logging.DEBUG if dev_mode else logging.INFO,
)
structlog.configure(
processors=[
structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S"), # Add timestamps
structlog.stdlib.add_log_level, # Add log levels
structlog.processors.StackInfoRenderer(), # Render stack info
structlog.processors.format_exc_info, # Format exceptions
# structlog.dev.RichTracebackFormatter(),
ConsoleRenderer(
# exception_formatter=structlog.dev.rich_traceback
)
if dev_mode
else JSONRenderer(), # Render logs nicely for dev
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
import logging
from fastapi_dynamic_response.settings import Settings
import structlog
def configure_logging_two():
settings = Settings()
# Clear existing loggers
logging.config.dictConfig(
{
"version": 1,
"disable_existing_loggers": False,
}
)
if settings.ENV == "local":
# Local development logging configuration
processors = [
# structlog.processors.TimeStamper(fmt="iso"),
structlog.dev.ConsoleRenderer(colors=False),
]
logging_level = logging.DEBUG
# Enable rich tracebacks
from rich.traceback import install
install(show_locals=True)
# Use RichHandler for pretty console logs
from rich.logging import RichHandler
logging.basicConfig(
level=logging_level,
format="%(message)s",
datefmt="[%X]",
handlers=[RichHandler()],
)
else:
# Production logging configuration
processors = [
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.JSONRenderer(),
]
logging_level = logging.INFO
# Standard logging configuration
logging.basicConfig(
format="%(message)s",
level=logging_level,
handlers=[logging.StreamHandler()],
)
structlog.configure(
processors=processors,
wrapper_class=structlog.make_filtering_bound_logger(logging_level),
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
cache_logger_on_first_use=True,
)
# Redirect uvicorn loggers to structlog
for logger_name in ("uvicorn", "uvicorn.error", "uvicorn.access"):
logger = logging.getLogger(logger_name)
logger.handlers = []
logger.propagate = True
configure_logging = configure_logging_two

24
src/fastapi_dynamic_response/main.py
View file

@ -7,14 +7,19 @@ from fastapi_dynamic_response.base.router import router as base_router
from fastapi_dynamic_response.dependencies import get_content_type
from fastapi_dynamic_response.middleware import (
Sitemap,
add_process_time_header,
catch_exceptions_middleware,
log_request_state,
log_requests,
respond_based_on_content_type,
set_prefers,
set_span_id,
)
from fastapi_dynamic_response.zpages.router import router as zpages_router
from fastapi_dynamic_response.logging_config import configure_logging
configure_logging()
app = FastAPI(
title="FastAPI Dynamic Response",
version=__version__,
@ -25,16 +30,23 @@ app = FastAPI(
# exception_handlers=exception_handlers,
debug=True,
dependencies=[
Depends(set_prefers),
Depends(set_span_id),
Depends(log_request_state),
# Depends(set_prefers),
# Depends(set_span_id),
# Depends(log_request_state),
],
)
# configure_tracing(app)
app.include_router(zpages_router)
app.include_router(base_router)
app.middleware("http")(Sitemap(app))
app.middleware("http")(catch_exceptions_middleware)
app.middleware("http")(respond_based_on_content_type)
app.middleware("http")(add_process_time_header)
app.middleware("http")(log_requests)
app.middleware("http")(Sitemap(app))
app.middleware("http")(set_prefers)
app.middleware("http")(set_span_id)
app.middleware("http")(catch_exceptions_middleware)
app.mount("/static", StaticFiles(directory="static"), name="static")

135
src/fastapi_dynamic_response/middleware.py
View file

@ -1,6 +1,7 @@
from difflib import get_close_matches
from io import BytesIO
import json
import time
import traceback
from typing import Any, Dict
from uuid import uuid4
@ -23,6 +24,10 @@ import base64
from fastapi_dynamic_response.constant import ACCEPT_TYPES
from fastapi_dynamic_response.globals import templates
import structlog
logger = structlog.get_logger()
console = Console()
@ -36,6 +41,13 @@ class Prefers(BaseModel):
png: bool = False
pdf: bool = False
def __repr__(self):
_repr = []
for key, value in self.dict().items():
if value:
_repr.append(key + "=True")
return f'Prefers({", ".join(_repr)})'
@property
def textlike(self) -> bool:
return self.rtf or self.text or self.markdown
@ -63,13 +75,29 @@ def log_request_state(request: Request):
console.log(request.state.prefers)
def set_span_id(request: Request):
async def add_process_time_header(request: Request, call_next):
start_time = time.perf_counter()
response = await call_next(request)
process_time = time.perf_counter() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
async def set_span_id(request: Request, call_next):
span_id = uuid4()
request.state.span_id = span_id
request.state.bound_logger = logger.bind(span_id=request.state.span_id)
response = await call_next(request)
response.headers["x-request-id"] = str(span_id)
response.headers["x-span-id"] = str(span_id)
return response
def set_prefers(
request: Request,
call_next,
):
content_type = (
request.query_params.get(
@ -93,24 +121,31 @@ def set_prefers(
user_agent = request.headers.get("user-agent", "").lower()
if hx_request_header == "true":
request.state.prefers = Prefers(html=True, partial=True)
return
content_type = "text/html-partial"
# request.state.prefers = Prefers(html=True, partial=True)
# content_type = "text/html"
if is_browser_request(user_agent) and content_type is None:
elif is_browser_request(user_agent) and content_type is None:
content_type = "text/html"
elif is_rtf_request(user_agent) and content_type is None:
content_type = "text/rtf"
elif content_type is None:
else:
content_type = "application/json"
partial = "partial" in content_type
# if content_type in ACCEPT_TYPES:
for accept_type, accept_value in ACCEPT_TYPES.items():
if accept_type in content_type:
request.state.prefers = Prefers(**{accept_value: True})
return
request.state.prefers = Prefers(JSON=True, partial=False)
request.state.prefers = Prefers(**{accept_value: True}, partial=partial)
request.state.content_type = content_type
request.state.bound_logger = request.state.bound_logger.bind(
# content_type=request.state.content_type,
prefers=request.state.prefers,
)
return call_next(request)
class Sitemap:
@ -318,7 +353,7 @@ async def respond_based_on_content_type(request: Request, call_next):
if str(response.status_code)[0] not in "123":
return response
return await handle_response(request, data)
return await handle_response(request, response, data)
# except TemplateNotFound:
# return HTMLResponse(content="Template Not Found ", status_code=404)
except StarletteHTTPException as exc:
@ -333,45 +368,55 @@ async def respond_based_on_content_type(request: Request, call_next):
return HTMLResponse(content=f"Internal Server Error: {e!s}", status_code=500)
async def handle_response(request: Request, data: str):
async def handle_response(request: Request, response: Response, data: str):
json_data = json.loads(data)
template_name = getattr(request.state, "template_name", "default_template.html")
if request.state.prefers.partial:
template_name = "partial_" + template_name
content_type = request.state.prefers
if request.state.prefers.JSON:
return JSONResponse(content=json_data)
elif request.state.prefers.html:
return templates.TemplateResponse(
template_name, {"request": request, "data": json_data}
return JSONResponse(
content=json_data,
)
elif request.state.prefers.markdown:
if request.state.prefers.html:
return templates.TemplateResponse(
template_name,
{"request": request, "data": json_data},
headers=response.headers,
)
if request.state.prefers.markdown:
import html2text
template = templates.get_template(template_name)
html_content = template.render(data=json_data)
markdown_content = html2text.html2text(html_content)
return PlainTextResponse(content=markdown_content)
return PlainTextResponse(content=markdown_content, headers=response.headers)
elif request.state.prefers.text:
if request.state.prefers.text:
plain_text_content = format_json_as_plain_text(json_data)
return PlainTextResponse(content=plain_text_content)
return PlainTextResponse(
content=plain_text_content,
)
elif request.state.prefers.rtf:
if request.state.prefers.rtf:
rich_text_content = format_json_as_rich_text(json_data, template_name)
return PlainTextResponse(content=rich_text_content)
return PlainTextResponse(
content=rich_text_content,
)
elif request.state.prefers.png:
if request.state.prefers.png:
template = templates.get_template(template_name)
html_content = template.render(data=json_data)
screenshot = get_screenshot(html_content)
return Response(content=screenshot.getvalue(), media_type="image/png")
return Response(
content=screenshot.getvalue(),
media_type="image/png",
)
elif request.state.prefers.pdf:
if request.state.prefers.pdf:
template = templates.get_template(template_name)
html_content = template.render(data=json_data)
scale = float(
@ -380,6 +425,42 @@ async def handle_response(request: Request, data: str):
console.log(f"Scale: {scale}")
pdf = get_pdf(html_content, scale)
return Response(content=pdf, media_type="application/pdf")
return Response(
content=pdf,
media_type="application/pdf",
)
return JSONResponse(content=json_data)
return JSONResponse(
content=json_data,
)
# Initialize the logger
async def log_requests(request: Request, call_next):
# Log request details
request.state.bound_logger.info(
"Request received",
# span_id=request.state.span_id,
method=request.method,
path=request.url.path,
# headers=dict(request.headers),
# prefers=request.state.prefers,
)
# logger.info(
# headers=dict(request.headers),
# prefers=request.state.prefers,
# )
# Process the request
response = await call_next(request)
# Log response details
# logger.info(
# "Response sent",
# span_id=request.state.span_id,
# method=request.method,
# status_code=response.status_code,
# headers=dict(response.headers),
# )
return response

8
src/fastapi_dynamic_response/settings.py Normal file
View file

@ -0,0 +1,8 @@
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
ENV: str = "local"
class Config:
env_file = "config.env"

35
src/fastapi_dynamic_response/tracing.py Normal file
View file

@ -0,0 +1,35 @@
# tracing.py
from fastapi_dynamic_response.settings import Settings
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
# from opentelemetry.exporter.richconsole import RichConsoleSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
def configure_tracing(app):
settings = Settings()
trace.set_tracer_provider(TracerProvider())
tracer_provider = trace.get_tracer_provider()
if settings.ENV == "local":
# Use console exporter for local development
# span_exporter = RichConsoleSpanExporter()
# span_processor = SimpleSpanProcessor(span_exporter)
# span_exporter = OTLPSpanExporter()
span_exporter = OTLPSpanExporter(
endpoint="http://localhost:4317", insecure=True
)
span_processor = BatchSpanProcessor(span_exporter)
else:
# Use OTLP exporter for production
span_exporter = OTLPSpanExporter()
span_processor = BatchSpanProcessor(span_exporter)
tracer_provider.add_span_processor(span_processor)
# Instrument FastAPI
FastAPIInstrumentor.instrument_app(app)