# src/sentry/conf/server.py:452-508
# Do not modify reordering
# The applications listed first in INSTALLED_APPS have precedence
INSTALLED_APPS: tuple[str, ...] = (
    "django.contrib.auth",              # Django 用户认证
    "django.contrib.contenttypes",      # ContentType 框架
    "django.contrib.humanize",          # 人性化显示过滤器
    "django.contrib.messages",          # Flash 消息系统
    "django.contrib.postgres",          # PostgreSQL 特有字段类型
    "django.contrib.sessions",          # 会话管理
    "django.contrib.sites",             # 多站点支持
    "drf_spectacular",                  # OpenAPI 文档生成
    "crispy_forms",                     # 表单渲染
    "rest_framework",                   # Django REST Framework
    "sentry",                           # 核心应用（数据模型、任务、工具函数）
    "sentry.analytics",                 # 分析事件追踪
    "sentry.auth_v2",                   # V2 认证前端
    "sentry.incidents.apps.Config",     # 告警事件管理
    "sentry.deletions",                 # 可级联删除框架
    "sentry.discover",                  # Discover 查询引擎
    "sentry.analytics.events",          # 分析事件类型定义
    "sentry.services.nodestore",        # Node 存储（事件数据）
    "sentry.users",                     # 用户管理
    "sentry.sentry_apps",               # Sentry App 平台
    "sentry.integrations",              # 第三方集成
    "sentry.notifications",             # 通知系统
    "sentry.flags",                     # Feature Flag 集成
    "sentry.monitors",                  # Cron 监控
    "sentry.processing_errors",         # 处理错误追踪
    "sentry.uptime",                    # 网站可用性监控
    "sentry.tempest",                   # Tempest 测试
    "sentry.ai_monitoring",             # AI/LLM 监控
    "sentry.replays",                   # Session Replay
    "sentry.release_health",            # 版本健康度
    "sentry.search",                    # 搜索后端
    "sentry.sentry_metrics",            # Metrics 指标系统
    "sentry.sentry_metrics.indexer.postgres.apps.Config",  # 指标索引
    "sentry.snuba",                     # Snuba 查询接口
    "social_auth",                      # OAuth 社交登录
    "sudo",                             # Sudo 模式
    "sentry.eventstream",               # 事件流
    "sentry.auth.providers.google.apps.Config",   # Google 登录
    "sentry.auth.providers.fly.apps.Config",      # Fly.io 登录
    "django.contrib.staticfiles",       # 静态文件服务
    "sentry.issues.apps.Config",        # Issues 模块
    "sentry.feedback",                  # 用户反馈
    "sentry.hybridcloud",               # 混合云基础设施
    "sentry.relocation",                # 数据迁移/搬迁
    "sentry.remote_subscriptions.apps.Config",    # 远程订阅
    "sentry.data_secrecy",              # 数据保密
    "sentry.workflow_engine",           # 工作流引擎
    "sentry.explore",                   # Explore 查询
    "sentry.insights",                  # Insights 洞察
    "sentry.preprod",                   # 预发布
    "sentry.releases",                  # 版本管理
    "sentry.seer",                      # Seer AI 引擎
    "sentry.scm",                       # 源代码管理
)

# src/sentry/conf/server.py:46-92
_env_cache: dict[str, object] = {}

@overload
def env(key: str) -> str: ...

@overload
def env(key: str, default: _EnvTypes, type: env_types.Type | None = None) -> _EnvTypes: ...

def env(
    key: str,
    default: str | _EnvTypes = "",
    type: env_types.Type | None = None,
) -> _EnvTypes:
    # 首先检查内部缓存
    try:
        rv = _env_cache[key]
    except KeyError:
        if "SENTRY_RUNNING_GRANIAN" in os.environ:
            fn = os.environ.pop    # Granian 模式下弹出变量（安全考虑）
        else:
            fn = os.environ.__getitem__  # 普通模式直接读取
        try:
            rv = fn(key)
            _env_cache[key] = rv
        except KeyError:
            rv = default

    if type is None:
        type = env_types.type_from_value(default)

    return type(rv)

# 布尔标志
SPOTLIGHT = IS_DEV and SPOTLIGHT_ENV_VAR.lower() not in ("0", "false", "n", "no")

# 字符串配置
SENTRY_BASE_HOSTNAME = env("SENTRY_BASE_HOSTNAME", "")

# 整数配置
SENTRY_API_RESPONSE_DELAY = env("SENTRY_API_RESPONSE_DELAY", 0, int)

# 列表配置
INTERNAL_IPS = env("SENTRY_INTERNAL_IPS", (), tuple)

SENTRY_OPTIONS: dict[str, Any] = {}
SENTRY_OPTIONS_COMPLAIN_ON_ERRORS = True

from sentry import options

# 读取
max_events = options.get("system.maximum-event-size")

# 写入（需要管理权限）
options.set("system.rate-limit", 1000)

# src/sentry/conf/urls.py
from __future__ import annotations

from sentry.web.urls import handler500, urlpatterns

__all__ = ("urlpatterns", "handler500")

# src/sentry/web/urls.py:69-84
urlpatterns = [
    re_path(r"^500/", Error500View.as_view(), name="error-500"),
    re_path(r"^404/", Error404View.as_view(), name="error-404"),
    re_path(r"^403-csrf-failure/", csrf_failure.view, name="error-403-csrf-failure"),
]

# src/sentry/web/urls.py:180-183
re_path(
    r"^api/0/",
    include("sentry.api.urls"),
),

# 大量路由都指向同一个 react_page_view
re_path(r"^issues/", react_page_view, name="issues"),
re_path(r"^alerts/", react_page_view, name="alerts"),
re_path(r"^performance/", react_page_view, name="performance"),
# ...

# /oauth/、/auth/、/account/、/saml/ 等
re_path(r"^oauth/", include([...])),
re_path(r"^auth/", include([...])),
re_path(r"^account/", include([...])),
re_path(r"^saml/", include([...])),

# /organizations/<slug>/... 下的各种页面
re_path(r"^organizations/", include([...])),

# /extensions/jira/、/extensions/slack/ 等
re_path(r"^extensions/", include([...])),

# 典型的路由注册模式（api/urls.py 样式）
re_path(
    r"^(?P<organization_id_or_slug>[^/]+)/projects/$",
    OrganizationProjectsEndpoint.as_view(),
    name="sentry-api-0-organization-projects",
),
re_path(
    r"^(?P<organization_id_or_slug>[^/]+)/projects/(?P<project_id_or_slug>[^/]+)/$",
    ProjectDetailsEndpoint.as_view(),
    name="sentry-api-0-project-details",
),

def create_group_urls(name_prefix: str) -> list[URLPattern | URLResolver]:
    return [
        re_path(r"^(?P<issue_id>[^/]+)/$",
            GroupDetailsEndpoint.as_view(),
            name=f"{name_prefix}-group-details"),
        re_path(r"^(?P<issue_id>[^/]+)/activities/$",
            GroupActivitiesEndpoint.as_view(), ...),
        re_path(r"^(?P<issue_id>[^/]+)/events/$",
            GroupEventsEndpoint.as_view(), ...),
        # ... 等 15+ 个 Issue 子资源路由
    ]

MIDDLEWARE: tuple[str, ...] = (
    "csp.middleware.CSPMiddleware",                         # 内容安全策略
    "sentry.middleware.health.HealthCheck",                 # 健康检查
    "sentry.middleware.security.SecurityHeadersMiddleware", # 安全响应头
    "sentry.middleware.env.SentryEnvMiddleware",            # 环境变量注入
    "sentry.middleware.proxy.SetRemoteAddrFromForwardedFor",# 代理 IP 处理
    "sentry.middleware.stats.RequestTimingMiddleware",      # 请求计时
    "sentry.middleware.access_log.access_log_middleware",   # 访问日志
    "sentry.middleware.stats.ResponseCodeMiddleware",       # 响应码统计
    "sentry.middleware.subdomain.SubdomainMiddleware",      # 子域名解析
    "django.middleware.common.CommonMiddleware",            # Django 通用中间件
    "django.contrib.sessions.middleware.SessionMiddleware", # Session 管理
    "django.middleware.csrf.CsrfViewMiddleware",            # CSRF 保护
    "sentry.middleware.auth.AuthenticationMiddleware",      # 认证中间件（自定义）
    "sentry.middleware.suspended.SuspendedUserMiddleware",  # 暂停用户检测
    "sentry.middleware.viewer_context.ViewerContextMiddleware", # Viewer 上下文
    "sentry.middleware.ai_agent.AIAgentMiddleware",         # AI Agent 检测
    "sentry.middleware.integrations.IntegrationControlMiddleware", # 集成控制
    APIGW_MIDDLEWARE,                                       # API 网关（动态选择）
    "sentry.middleware.demo_mode_guard.DemoModeGuardMiddleware",   # Demo 模式
    "sentry.middleware.customer_domain.CustomerDomainMiddleware",  # 客户域名
    "sentry.middleware.sudo.SudoMiddleware",                # Sudo 模式
    "sentry.middleware.superuser.SuperuserMiddleware",      # 超级用户
    "sentry.middleware.staff.StaffMiddleware",              # Staff 用户
    "sentry.middleware.locale.SentryLocaleMiddleware",      # 国际化
    "sentry.middleware.ratelimit.RatelimitMiddleware",      # 速率限制
    "django.contrib.messages.middleware.MessageMiddleware", # Flash 消息
    "sentry.middleware.devtoolbar.DevToolbarAnalyticsMiddleware",  # 工具栏分析
    "sentry.middleware.agent_discovery.AgentDiscoveryMiddleware", # Agent 发现
)

# src/sentry/middleware/auth.py:1-60
class AuthenticationMiddleware(MiddlewareMixin):
    def process_request(self, request: Request) -> None:
        if request.path.startswith("/api/0/internal/rpc/"):
            # RPC 请求跳过认证（由 RPC 层自行处理）
            request.user, request.auth = AnonymousUser(), None
        # ... 其余逻辑

# src/sentry/conf/server.py:383-389
APIGW_ASYNC = os.environ.get("SENTRY_APIGW_ASYNC", "").lower() in ("1", "true", "y", "yes")
APIGW_MIDDLEWARE = (
    "sentry.hybridcloud.apigateway_async.middleware.ApiGatewayMiddleware"
    if APIGW_ASYNC
    else "sentry.hybridcloud.apigateway.middleware.ApiGatewayMiddleware"
)

# src/sentry/api/base.py:143-196
def apply_cors_headers(request, response, allowed_methods=None):
    # ...
    response["Access-Control-Allow-Headers"] = (
        "X-Sentry-Auth, X-Requested-With, Origin, Accept, "
        "Content-Type, Authentication, Authorization, Content-Encoding, "
        "sentry-trace, baggage, X-CSRFToken"
    )
    response["Access-Control-Expose-Headers"] = (
        "X-Sentry-Error, X-Sentry-Direct-Hit, X-Hits, X-Max-Hits, "
        "Endpoint, Retry-After, Link"
    )
    # ...

# src/sentry/api/base.py:242-243
class Endpoint(APIView):
    rate_limits: RateLimitConfig | Callable[..., RateLimitConfig] = DEFAULT_RATE_LIMIT_CONFIG
    enforce_rate_limit: bool = settings.SENTRY_RATELIMITER_ENABLED

class Endpoint(APIView):
    authentication_classes: tuple[type[BaseAuthentication], ...] = DEFAULT_AUTHENTICATION
    permission_classes: tuple[type[BasePermission], ...] = (NoPermission,)

    cursor_name = "cursor"

    owner: ApiOwner = ApiOwner.UNOWNED
    publish_status: dict[HTTP_METHOD_NAME, ApiPublishStatus] = {}
    rate_limits: RateLimitConfig | Callable[..., RateLimitConfig] = DEFAULT_RATE_LIMIT_CONFIG
    enforce_rate_limit: bool = settings.SENTRY_RATELIMITER_ENABLED

# src/sentry/api/base.py:382-495（关键逻辑摘要）
@csrf_exempt
@allow_cors_options
def dispatch(self, request, *args, **kwargs):
    # 1. 初始化请求（补丁 auth/user 用于内部 API 客户端）
    request = self.initialize_request(request, *args, **kwargs)
    request.body  # 预读 body

    # 2. Origin 校验
    if origin:
        if not is_valid_origin(origin, allowed=allowed_origins):
            return Response(f"Invalid origin: {origin}", status=400)

    # 3. Token 最后使用时间更新
    if request.auth:
        update_token_access_record(request.auth)

    # 4. DRF 初始检查（认证、权限、节流）
    self.initial(request, *args, **kwargs)

    # 5. 自动设置请求级 access 对象（Access 权限模型）
    if getattr(request, "access", None) is None:
        request.access = access.from_request(request)

    # 6. 参数转换 + 调用具体 handler（get/post/put/delete）
    (args, kwargs) = self.convert_args(request, *args, **kwargs)
    response = handler(request, *args, **kwargs)

    # 7. 开发环境分页强制检查
    if settings.ENFORCE_PAGINATION:
        # 如果 GET 方法返回数组且未分页，抛出 MissingPaginationError
        ...

class NoPermission(BasePermission):
    def has_permission(self, request, view):
        return False        # 默认拒绝一切

class SuperuserPermission(BasePermission):
    def has_permission(self, request, view):
        return is_active_superuser(request)   # 需要激活的超级用户

class StaffPermission(BasePermission):
    def has_permission(self, request, view):
        return is_active_staff(request)       # 需要 Staff 成员

class SystemPermission(BasePermission):
    def has_permission(self, request, view):
        return is_system_auth(request.auth)   # 需要系统级 Token

# 典型的端点权限声明
from sentry.api.bases.organization import OrganizationPermission

class OrganizationProjectsEndpoint(Endpoint):
    permission_classes = (OrganizationPermission,)

def paginate(
    self, request,
    on_results=None,
    paginator=None,
    paginator_cls=Paginator,
    default_per_page=None,
    max_per_page=None,
    cursor_cls=Cursor,
    response_cls=Response,
    response_kwargs=None,
    count_hits=None,
    **paginator_kwargs,
):
    per_page = self.get_per_page(request, default_per_page, max_per_page)
    cursor = self.get_cursor_from_request(request, cursor_cls)
    paginator = get_paginator(paginator, paginator_cls, paginator_kwargs)
    cursor_result = paginator.get_result(limit=per_page, cursor=cursor, ...)

    # 可选的结果映射
    if on_results:
        results = on_results(cursor_result.results)
    else:
        results = cursor_result.results

    response = response_cls(results, **response_kwargs)
    self.add_cursor_headers(request, response, cursor_result)
    return response

# src/sentry/api/base.py:606-670
class StatsMixin:
    def _parse_args(self, request, environment_id=None, restrict_rollups=True):
        resolution_s = request.GET.get("resolution")
        # ... 解析 since、until 时间戳
        # ... 自动选择最优 rollup
        return {
            "start": start,
            "end": end,
            "rollup": resolution,
            "environment_ids": environment_id and [environment_id],
        }

# 典型端点结构
class OrganizationProjectsEndpoint(Endpoint):
    owner = ApiOwner.ISSUES
    publish_status = {
        "GET": ApiPublishStatus.PUBLIC,
        "POST": ApiPublishStatus.PUBLIC,
    }
    permission_classes = (OrganizationPermission,)

    def get(self, request, organization):
        # 处理 GET 请求
        project_list = Project.objects.filter(organization=organization)
        return self.paginate(
            request=request,
            queryset=project_list,
            on_results=lambda x: serialize(x, request.user),
        )

    def post(self, request, organization):
        # 处理 POST 请求
        serializer = ProjectPostSerializer(data=request.data)
        if not serializer.is_valid():
            return self.respond(serializer.errors, status=400)
        project = serializer.save()
        return self.respond(serialize(project, request.user), status=201)

from sentry.api.base import control_silo_endpoint, cell_silo_endpoint, all_silo_endpoint

@control_silo_endpoint     # 仅在 CONTROL Silo 可用
class OrganizationIndexEndpoint(Endpoint):
    ...

@cell_silo_endpoint        # 仅在 CELL Silo 可用
class ProjectDetailsEndpoint(Endpoint):
    ...

@all_silo_endpoint         # 所有 Silo 均可用（稀少）
class HealthCheckEndpoint(Endpoint):
    ...

# src/sentry/api/serializers/base.py:145-216
class Serializer(Generic[T]):
    def __call__(self, obj, attrs, user, **kwargs):
        # 如果 attrs 非空（已通过 get_attrs 批量预取），使用 attrs
        # 否则调用 serialize 方法
        return self.serialize(obj, user, **kwargs)

    def get_attrs(self, item_list, user, **kwargs):
        # 批量预取/聚合数据，对所有待序列化的对象一次性完成
        # 子类重写此方法以优化查询（N+1 问题）
        return {}

    def serialize(self, obj, user, **kwargs):
        raise NotImplementedError

@register(Organization)
class OrganizationSerializer(Serializer):
    def serialize(self, obj, attrs, user, **kwargs):
        return {
            "id": str(obj.id),
            "slug": obj.slug,
            "name": obj.name,
            # ...
        }

def serialize(objects, user=None, serializer=None, **kwargs):
    if not objects:
        return objects
    elif not isinstance(objects, (list, tuple, set, frozenset)):
        return serialize([objects], user=user, serializer=serializer, **kwargs)[0]

    if serializer is None:
        for o in objects:
            try:
                serializer = registry[type(o)]
                break
            except KeyError:
                pass
        else:
            return objects  # 未找到序列化器，返回原对象

    # 阶段 1：批量获取属性
    attrs = serializer.get_attrs(
        item_list=[o for o in objects if o is not None],
        user=user, **kwargs,
    )

    # 阶段 2：逐个序列化
    return [serializer(o, attrs=attrs.get(o, {}), user=user, **kwargs) for o in objects]

def get(self, request, organization):
    projects = Project.objects.filter(organization=organization)
    # 带 ProjectSummarySerializer 的序列化，响应类型为 list[ProjectSummary]
    return self.respond(serialize(
        list(projects),
        request.user,
        ProjectSummarySerializer(),
    ))

# src/sentry/api/serializers/rest_framework/base.py
from rest_framework.serializers import ModelSerializer, Serializer

# src/sentry/api/serializers/rest_framework/base.py:51-63
def camel_to_snake_case(value):
    """CamelCase -> snake_case"""
    return re_camel_case.sub(r"_\1", value).strip("_").lower()

def snake_to_camel_case(value):
    """snake_case -> camelCase"""
    words = value.strip("_").split("_")
    return words[0].lower() + "".join(word.capitalize() for word in words[1:])

# 度量记录，用于监控 API 使用习惯
def _record_key_case_metric(serializer_name, data):
    key_case = _classify_key_case(data)  # camel / snake / mixed / uncertain
    metrics.incr("api.serializer.parameter_key_case", tags={...})

# src/sentry/api/base.py:106-113
DEFAULT_AUTHENTICATION = (
    UserAuthTokenAuthentication,
    OrgAuthTokenAuthentication,
    AgentTokenAuthentication,
    ApiKeyAuthentication,
    ViewerContextAuthentication,
    SessionAuthentication,
)

# src/sentry/api/authentication.py（逻辑摘要）
class UserAuthTokenAuthentication(BaseAuthentication):
    def authenticate(self, request):
        header = get_authorization_header(request)
        token_str = parse_bearer_token(header)
        # 1. 哈希 Token 并查询数据库
        token = ApiToken.objects.filter(
            token=hash_token(token_str), expires_at__gt=now()
        ).first()
        # 2. 验证 Token 是否过期、是否被吊销
        # 3. 加载关联用户并检查用户状态
        # 4. 在 request.auth 上设置 Token 的 scope 列表
        return (user, token)

# src/sentry/middleware/auth.py:24-52
def get_user(request):
    if not hasattr(request, "_cached_user"):
        user = auth_get_user(request)
        # 验证 session nonce
        if user.is_authenticated and not user.is_sentry_app:
            if user.session_nonce and request.session.get("_nonce", "") != user.session_nonce:
                # nonce 不匹配，会话失效
                user = AnonymousUser()
            else:
                UserIP.log(user, request.META["REMOTE_ADDR"])
        request._cached_user = user
    return request._cached_user

class AuthenticationSiloLimit(SiloLimit):
    def handle_when_unavailable(self, original_method, current_mode, available_modes):
        def handle(obj, *args, **kwargs):
            message = (
                f"{type(obj)} used for an endpoint in {current_mode} mode. "
                f"This authenticator is available only in: {mode_str}"
            )
            raise self.AvailabilityError(message)
        return handle

class SentryAPIException(APIException):
    code = ""
    message = ""

    def __init__(self, code=None, message=None, detail=None, **kwargs):
        if detail is None:
            detail = {
                "code": code or self.code,
                "message": message or self.message,
                "extra": kwargs,
            }
        self.detail = {"detail": detail}

class InsufficientScope(PermissionDenied):
    def __init__(self, required_scopes):
        super().__init__()
        scope = " ".join(sorted(required_scopes))
        self.auth_header = f'Bearer error="insufficient_scope", scope="{scope}"'

def handle_exception_with_details(self, request, exc, handler_context=None, scope=None):
    try:
        # 首先尝试 DRF 内置异常处理器
        # 如果 settings.EXCEPTION_HANDLER 存在且返回响应，直接使用
        response = self.handle_exception(exc)
    except Exception as err:
        # DRF 也无法处理的异常 → 捕获到 Sentry 自身
        import traceback
        sys.stderr.write(traceback.format_exc())
        scope = scope or sentry_sdk.Scope()
        if handler_context:
            merge_context_into_scope("Request Handler Data", handler_context, scope)
        event_id = capture_exception(err, scope=scope)

        response_body = {"detail": "Internal Error", "errorId": event_id}
        response = Response(response_body, status=500)
        response.exception = True
    return response

from drf_spectacular.utils import extend_schema, extend_schema_serializer

@extend_schema_serializer(exclude_fields=["internal_field"])
class OrganizationPostSerializer(BaseOrganizationSerializer):
    ...

@extend_schema(
    parameters=[OrganizationParams, CursorQueryParam],
    responses={
        200: inline_sentry_response_serializer("OrganizationList", list[OrgResponse]),
        403: RESPONSE_FORBIDDEN,
    },
)
class OrganizationIndexEndpoint(Endpoint):
    ...

inline_sentry_response_serializer(
    "OrganizationList",
    list[OrganizationSummarySerializerResponse]
)