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

def dispatch(self, request: Request, *args, **kwargs) -> Response:

def convert_args(self, request: Request, *args, **kwargs):
    return (args, kwargs)

def handle_exception_with_details(self, request, exc, ...):
    response = self.handle_exception(exc)
    ...
    response_body = {"detail": "Internal Error", "errorId": event_id}
    return Response(response_body, status=500)

class OrganizationEndpoint(Endpoint):
    permission_classes: tuple[type[BasePermission], ...] = (OrganizationPermission,)

def get_projects(
    self, request, organization, *,
    force_global_perms=False, include_all_accessible=False,
    project_ids=None, project_slugs=None,
) -> list[Project]:

class ProjectEndpoint(Endpoint):
    permission_classes: tuple[type[BasePermission], ...] = (ProjectPermission,)

try:
    redirect = ProjectRedirect.objects.select_related("project").get(
        organization__slug__id_or_slug=organization_id_or_slug,
        redirect_slug=project_id_or_slug,
    )
    self.check_object_permissions(request, redirect.project)
    # 构造新 URL 并返回 302
    raise ProjectMoved(new_url, redirect.project.slug)
except ProjectRedirect.DoesNotExist:
    raise ProjectDoesNotExist

class TeamEndpoint(Endpoint):
    permission_classes: tuple[type[BasePermission], ...] = (TeamPermission,)

# Cell Silo 端点（最常用）
@cell_silo_endpoint
class OrganizationProjectsSentFirstEventEndpoint(OrganizationEndpoint):
    ...

# Control Silo 端点
@control_silo_endpoint
class SomeControlEndpoint(Endpoint):
    ...

# 所有 Silo 可用（极少数情况）
@all_silo_endpoint
class RobotsTxtEndpoint(Endpoint):
    ...

class ScopedPermission(BasePermission):
    scope_map: dict[str, Sequence[str]] = {
        "HEAD": (),
        "GET": (),
        "POST": (),
        "PUT": (),
        "PATCH": (),
        "DELETE": (),
    }

def determine_access(self, request, organization):
    org_context = organization_service.get_organization_by_id(...)
    if request.auth:
        request.access = access.from_request_org_and_scopes(
            request=request, rpc_user_org_context=org_context,
            scopes=request.auth.get_scopes(),
        )
    else:
        request.access = access.from_request_org_and_scopes(
            request=request, rpc_user_org_context=org_context,
        )
    # 然后检查 2FA、SSO、成员限额

class OrganizationPermission(DemoSafePermission):
    scope_map = {
        "GET": ["org:read", "org:write", "org:admin"],
        "POST": ["org:write", "org:admin"],
        "PUT": ["org:write", "org:admin"],
        "DELETE": ["org:admin"],
    }

class ProjectPermission(OrganizationPermission):
    scope_map = {
        "GET": ["project:read", "project:write", "project:admin"],
        "POST": ["project:write", "project:admin"],
        "PUT": ["project:write", "project:admin"],
        "DELETE": ["project:admin"],
    }

    def has_object_permission(self, request, view, project):
        has_org_scope = super().has_object_permission(request, view, project.organization)
        if has_org_scope and request.access.has_project_access(project):
            return has_org_scope
        allowed_scopes = set(self.scope_map.get(request.method, []))
        return request.access.has_any_project_scope(project, allowed_scopes)

class TeamPermission(OrganizationPermission):
    scope_map = {
        "GET": ["team:read", "team:write", "team:admin"],
        "POST": ["team:write", "team:admin"],
        "PUT": ["team:write", "team:admin"],
        "DELETE": ["team:admin"],
    }

class CamelSnakeModelSerializer(ModelSerializer):
    def __init__(self, instance=None, data=empty, **kwargs):
        if data is not empty:
            if isinstance(data, dict):
                _record_key_case_metric(type(self).__name__, data)
            data = convert_dict_key_case(data, camel_to_snake_case)
        super().__init__(instance=instance, data=data, **kwargs)

    @property
    def errors(self):
        errors = super().errors
        return convert_dict_key_case(errors, snake_to_camel_case)

class ProjectField(serializers.Field):
    def __init__(self, scope="project:write", id_allowed=False, **kwargs):
        self.scope = scope
        self.id_allowed = id_allowed

    def to_internal_value(self, data):
        project = Project.objects.get(
            organization=self.context["organization"],
            slug__id_or_slug=project_id_or_slug
        )
        if not self.context["access"].has_any_project_scope(project, scopes):
            raise ValidationError("Insufficient access to project")
        return project

# 布尔标记
include_values_seen = request.GET.get("includeValuesSeen") != "0"
only_sampling_tags = request.GET.get("onlySamplingTags") == "1"

# 列表参数
project_ids = request.GET.getlist("project")

# 分页参数
per_page = self.get_per_page(request, default_per_page=100, max_per_page=100)
cursor = self.get_cursor_from_request(request)

# URL: /api/0/organizations/{organization_id_or_slug}/projects/
def get(self, request: Request, organization: Organization) -> Response:
    # organization 已经通过 convert_args 从数据库获取并经过权限检查
    projects = self.get_projects(request, organization)
    ...

# 单条资源
def get(self, request, project, rule):
    return Response(serialize(rule, request.user))

# 自定义结构化响应
def get(self, request, organization):
    return Response(serialize({"sentFirstEvent": seen_first_event}, request.user))

def respond(self, context: object | None = None, **kwargs: Any) -> Response:
    return Response(context, **kwargs)

def paginate(self, request, on_results=None, paginator=None,
             paginator_cls=Paginator, default_per_page=None,
             max_per_page=None, cursor_cls=Cursor, ...):

per_page = self.get_per_page(request, default_per_page=100, max_per_page=100)
cursor = self.get_cursor_from_request(request)

class ResourceDoesNotExist(APIException):
    status_code = status.HTTP_404_NOT_FOUND
    default_detail = "The requested resource does not exist"

class Serializer(Generic[T]):
    def get_attrs(self, item_list, user, **kwargs) -> MutableMapping[Any, Any]:
        return {}

    def __call__(self, obj, attrs, user, **kwargs) -> T:
        return self.serialize(obj, attrs, user, **kwargs)

    def serialize(self, obj, attrs, user, **kwargs):
        return {}

def serialize(objects, user=None, serializer=None, **kwargs):
    # 1. 单个对象自动包装为列表处理
    # 2. 通过注册表查找匹配的 Serializer（如果未显式传递）
    # 3. 调用 serializer.get_attrs(item_list, user, **kwargs) 批量获取属性
    # 4. 逐个调用 serializer(obj, attrs=attrs.get(obj, {}), user=user, **kwargs)
    ...

@register(Project)
class ProjectSerializer(Serializer):
    ...

@register(Project)
class ProjectSerializer(Serializer):
    def get_attrs(self, item_list, user, **kwargs):
        # 批量预取关联数据
        project_ids = [p.id for p in item_list]
        ...
        # 批量查询平台信息
        platforms = ProjectPlatform.objects.filter(project_id__in=project_ids)
        ...
        return {
            p: {
                "platform": platforms_by_project.get(p.id),
                "bookmark": bookmarks_by_project.get(p.id),
                "team": team_list_by_project.get(p.id, []),
                ...
            }
            for p in item_list
        }

    def serialize(self, obj, attrs, user, **kwargs):
        return {
            "id": str(obj.id),
            "slug": obj.slug,
            "name": obj.name,
            "platform": attrs.get("platform"),
            "isBookmarked": attrs.get("bookmark") is not None,
            "teams": [{"id": str(t.id), "name": t.name, "slug": t.slug}
                       for t in attrs.get("team", [])],
            ...
        }

from sentry.api.serializers import serialize
from sentry.api.serializers.models.team import TeamSerializer

def serialize(self, obj, attrs, user, **kwargs):
    return {
        ...
        "organization": serialize(obj.organization, user),
        "teams": serialize(attrs["team_list"], user, TeamSerializer()),
    }

collapse = kwargs.get("collapse", [])

data = { ... }
if "organization" not in collapse:
    data["organization"] = serialize(attrs["org"], user)
if "unusedFeatures" in collapse:
    # 排除前端不使用的 features，减少响应体积
    features = [f for f in features if f not in UNUSED_FEATURES]

class ProjectTagKeyResponse(TypedDict):
    key: str
    name: str
    canDelete: bool
    uniqueValues: NotRequired[int]

@extend_schema(
    responses={
        200: inline_sentry_response_serializer(
            "ListProjectTagKeys", list[ProjectTagKeyResponse]
        ),
    },
)

# 此端点响应为匿名结构，无需定义单独的 TypedDict
# 直接在 handler 中构造响应内容

from sentry.api.serializers import serialize

# 端点不涉及复杂的模型序列化，直接使用 serialize() 函数
# 传入 dict 对象时会经过序列化处理

# 文件: src/sentry/api/endpoints/organization_projects_sent_first_event.py

from rest_framework.request import Request
from rest_framework.response import Response

from sentry.api.api_owners import ApiOwner
from sentry.api.api_publish_status import ApiPublishStatus
from sentry.api.base import cell_silo_endpoint
from sentry.api.bases.organization import OrganizationEndpoint
from sentry.api.serializers import serialize
from sentry.models.organization import Organization


@cell_silo_endpoint
class OrganizationProjectsSentFirstEventEndpoint(OrganizationEndpoint):
    owner = ApiOwner.ISSUES
    publish_status = {
        "GET": ApiPublishStatus.PRIVATE,
    }

    def get(self, request: Request, organization: Organization) -> Response:
        """Verify If Any Project Within An Organization Has Received a First Event

        Returns true if any projects within the organization have received
        a first event, false otherwise.
        """
        projects = self.get_projects(request, organization)
        seen_first_event = any(p.first_event for p in projects)
        return Response(serialize({"sentFirstEvent": seen_first_event}, request.user))

from sentry.api.endpoints.organization_projects_sent_first_event import (
    OrganizationProjectsSentFirstEventEndpoint,
)

# 在组织级路由组中添加
re_path(
    r"^(?P<organization_id_or_slug>[^/]+)/projects-sent-first-event/$",
    OrganizationProjectsSentFirstEventEndpoint.as_view(),
    name="sentry-api-0-organization-sent-first-event",
),

# 文件: tests/sentry/api/endpoints/test_organization_projects_sent_first_event.py

from datetime import UTC, datetime
from django.urls import reverse
from sentry.testutils.cases import APITestCase


class OrganizationProjectsSentFirstEventEndpointTest(APITestCase):
    def setUp(self):
        self.foo = self.create_user("foo@example.com")
        self.org = self.create_organization(owner=self.user)
        self.team = self.create_team(organization=self.org)
        self.url = reverse(
            "sentry-api-0-organization-sent-first-event",
            kwargs={"organization_id_or_slug": self.org.slug},
        )

    def test_simple_sent_first_event(self):
        self.create_project(teams=[self.team], first_event=datetime.now(UTC))
        self.create_member(organization=self.org, user=self.foo, teams=[self.team])
        self.login_as(user=self.foo)

        response = self.client.get(self.url)
        assert response.status_code == 200
        assert response.data["sentFirstEvent"]

    def test_simple_no_first_event(self):
        self.create_project(teams=[self.team])
        self.create_member(organization=self.org, user=self.foo, teams=[self.team])
        self.login_as(user=self.foo)

        response = self.client.get(self.url)
        assert response.status_code == 200
        assert not response.data["sentFirstEvent"]

    def test_first_event_from_project_ids(self):
        project = self.create_project(teams=[self.team], first_event=datetime.now(UTC))
        self.create_member(organization=self.org, user=self.foo)
        self.login_as(user=self.foo)

        response = self.client.get(f"{self.url}?project={project.id}")
        assert response.status_code == 200
        assert response.data["sentFirstEvent"]

    def test_no_first_event_in_member_projects(self):
        """用户不是项目团队成员时，get_projects 过滤后列表为空"""
        self.create_project(teams=[self.team], first_event=datetime.now(UTC))
        self.create_member(organization=self.org, user=self.foo)  # 未加入团队
        self.login_as(user=self.foo)

        response = self.client.get(self.url)
        assert response.status_code == 200
        assert not response.data["sentFirstEvent"]

name="sentry-api-0-organization-sent-first-event"

reverse("sentry-api-0-organization-sent-first-event",
        kwargs={"organization_id_or_slug": self.org.slug})

class ApiPublishStatus(Enum):
    PUBLIC = "public"          # 稳定 API，出现在公开文档中
    PRIVATE = "private"        # 内部 API，不对外发布
    EXPERIMENTAL = "experimental"  # 开发中的 API，未来将公开

publish_status = {
    "GET": ApiPublishStatus.PUBLIC,
    "POST": ApiPublishStatus.PRIVATE,
}

class ApiOwner(Enum):
    ISSUES = "issue-workflow"
    REPLAY = "replay-backend"
    ALERTS_MONITORS = "alerts-monitors"
    HYBRID_CLOUD = "hybrid-cloud"
    UNOWNED = "unowned"
    # ... 共 22 个团队

from drf_spectacular.utils import extend_schema

@extend_schema(tags=["Projects"])
@cell_silo_endpoint
class ProjectTagsEndpoint(ProjectEndpoint):
    @extend_schema(
        operation_id="List a Project's Tag Keys",
        parameters=[
            GlobalParams.ORG_ID_OR_SLUG,
            GlobalParams.PROJECT_ID_OR_SLUG,
            OpenApiParameter(
                name="includeValuesSeen",
                location="query",
                required=False,
                type=str,
                description="Set to `0` to omit the `uniqueValues` count for each tag key.",
            ),
        ],
        responses={
            200: inline_sentry_response_serializer(
                "ListProjectTagKeys", list[ProjectTagKeyResponse]
            ),
            401: RESPONSE_UNAUTHORIZED,
            403: RESPONSE_FORBIDDEN,
            404: RESPONSE_NOT_FOUND,
        },
    )
    def get(self, request, project):
        ...

GlobalParams.ORG_ID_OR_SLUG        # 组织 ID/Slug 路径参数
GlobalParams.PROJECT_ID_OR_SLUG    # 项目 ID/Slug 路径参数
CursorQueryParam                   # 分页游标查询参数

RESPONSE_UNAUTHORIZED   # 401
RESPONSE_FORBIDDEN      # 403
RESPONSE_NOT_FOUND      # 404
RESPONSE_ACCEPTED       # 202

inline_sentry_response_serializer("ListProjectTagKeys", list[ProjectTagKeyResponse])

@extend_schema(
    examples=IssueAlertExamples.GET_PROJECT_RULE,
)

from sentry.testutils.cases import APITestCase
from django.urls import reverse

class MyEndpointTest(APITestCase):
    def setUp(self):
        self.login_as(self.user)  # 模拟登录
        self.url = reverse("sentry-api-0-my-endpoint",
                           kwargs={"organization_id_or_slug": self.organization.slug})

    def test_get_success(self):
        response = self.client.get(self.url)
        assert response.status_code == 200
        assert "expected_key" in response.data

def test_requires_authentication(self):
    """未认证用户应返回 401"""
    # 不调用 login_as
    response = self.client.get(self.url)
    assert response.status_code == 401

def test_non_member_denied(self):
    """非组织成员应返回 403"""
    non_member = self.create_user("outsider@example.com")
    self.login_as(non_member)
    response = self.client.get(self.url)
    assert response.status_code == 403  # 或 404（隐藏资源存在性）

def test_insufficient_scope(self):
    """Token scope 不足应返回 403"""
    token = self.create_api_token(self.user, scopes=["project:read"])  # 缺少 org:read
    response = self.client.get(self.url, HTTP_AUTHORIZATION=f"Bearer {token.token}")
    assert response.status_code == 403

def test_superuser_can_access(self):
    """超级用户可以访问"""
    superuser = self.create_user("admin@example.com", is_superuser=True)
    self.login_as(superuser)
    response = self.client.get(self.url)
    assert response.status_code == 200

# 创建组织
org = self.create_organization(owner=self.user)

# 创建团队
team = self.create_team(organization=org)

# 创建项目
project = self.create_project(organization=org, teams=[team])

# 创建成员
member = self.create_member(
    organization=org, user=other_user,
    role="member", teams=[team]
)

# 创建 API Token
token = self.create_api_token(user, scopes=["org:read"])

# 模型工厂方法
event = self.store_event(data={}, project_id=project.id)
rule = self.create_project_rule(project=project)
key = self.create_project_key(project=project)

from django.core.cache import cache

# OrganizationEndpoint 中的组织对象缓存
Organization.objects.get_from_cache(slug=organization_id_or_slug)

# OrganizationReleasesBaseEndpoint 中的发布权限缓存（60秒）
key = "release_perms:1:%s" % hash_values([...])
has_perms = cache.get(key)
if has_perms is None:
    has_perms = ...  # 计算权限
    cache.set(key, has_perms, 60)

(start, end) = clamp_date_range(
    default_start_end_dates(),
    timedelta(days=options.get("visibility.tag-key-max-date-range.days")),
)

# _validate_fetched_projects 确保用户请求的项目都在其权限范围内
if requesting_specific_projects:
    _validate_fetched_projects(filtered_projects, slugs, ids)

# 发布操作前检查发布的项目是否属于当前组织
projects = self.get_projects(request, organization, project_ids=project_ids)
ReleaseProject.objects.filter(release=release, project__in=projects).exists()

@dataclass(frozen=True)
class RateLimitConfig:
    group: str = "default"          # 速率限制组，如 "CLI"、"INTERNAL"
    limit_overrides: ...            # 按 HTTP 方法 + 类别的覆盖配置

class MyEndpoint(OrganizationEndpoint):
    rate_limits = RateLimitConfig(
        group="CLI",
        limit_overrides={
            "GET": {
                RateLimitCategory.ORGANIZATION: RateLimit(100, 1),  # 100次/秒
            },
        },
    )