1  import logging
 2  import sys
 3  
 4  import sentry_sdk
 5  from fastapi import FastAPI
 6  from sentry_sdk.integrations.aiohttp import AioHttpIntegration
 7  from sentry_sdk.integrations.logging import LoggingIntegration, ignore_logger
 8  from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration
 9  from uvicorn.config import LOGGING_CONFIG
10  from uvicorn.logging import DefaultFormatter
11  
12  from .settings import settings
13  
14  
15  def setup_sentry(app: FastAPI, dsn: str, name: str, version: str) -> None:
16      """Initialize sentry connection."""
17  
18      sentry_sdk.init(
19          dsn=dsn,
20          attach_stacktrace=True,
21          shutdown_timeout=5,
22          integrations=[
23              AioHttpIntegration(),
24              SqlalchemyIntegration(),
25              LoggingIntegration(level=logging.DEBUG, event_level=logging.WARNING),
26          ],
27          release=f"{name}@{version}",
28      )
29      ignore_logger("uvicorn.error")
30  
31  
32  logging_formatter = DefaultFormatter(fmt := "[%(asctime)s] %(levelprefix)s %(message)s")
33  LOGGING_CONFIG["formatters"]["default"]["fmt"] = fmt
34  LOGGING_CONFIG["formatters"]["access"][
35      "fmt"
36  ] = '[%(asctime)s] %(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s'
37  
38  logging_handler = logging.StreamHandler(sys.stdout)
39  logging_handler.setFormatter(logging_formatter)
40  
41  
42  def get_logger(name: str) -> logging.Logger:
43      """Get a logger with a given name."""
44  
45      logger: logging.Logger = logging.getLogger(name)
46      logger.addHandler(logging_handler)
47      logger.setLevel(settings.log_level)
48  
49      return logger