Undo sample changes

Author: adamtheturtle

sample/index.rst

@@ -106,287 +106,16 @@ This is **bold** and *italic* and ``inline code``.
 
 .. code-block:: python
 
-   from collections import deque
-   from typing import Protocol
-
-
-   class Limit(Protocol):
-      limit_id: str
-
-      def is_allowed(
-         self,
-         user_id: int,
-         request_url: str,
-         request_body: bytes,
-         request_timestamp: float,
-      ) -> bool: ...
-
-      def record_request(
-         self,
-         user_id: int,
-         request_url: str,
-         request_body: bytes,
-         request_timestamp: float,
-      ) -> None: ...
-
-
-   class RequestCountLimit:
-      def __init__(
-         self,
-         limit_id: str,
-         urls: list[str],
-         max_requests: int,
-         window_seconds: float,
-      ) -> None:
-         self.limit_id = limit_id
-         self.urls = urls
-         self.max_requests = max_requests
-         self.window_seconds = window_seconds
-         self.user_requests: dict[int, deque[float]] = {}
-
-      def is_allowed(
-         self,
-         user_id: int,
-         request_url: str,
-         request_body: bytes,
-         request_timestamp: float,
-      ) -> bool:
-         del request_body
-         if request_url not in self.urls:
-               return True
-
-         if user_id not in self.user_requests:
-               self.user_requests[user_id] = deque()
-
-         requests = self.user_requests[user_id]
-         cutoff_time = request_timestamp - self.window_seconds
-
-         while requests and requests[0] <= cutoff_time:
-               requests.popleft()
-
-         return len(requests) < self.max_requests
-
-      def record_request(
-         self,
-         user_id: int,
-         request_url: str,
-         request_body: bytes,
-         request_timestamp: float,
-      ) -> None:
-         del request_body
-         if request_url not in self.urls:
-               return
-
-         self.user_requests[user_id].append(request_timestamp)
-
-
-   class BodySizeLimit:
-      def __init__(
-         self,
-         limit_id: str,
-         url: str,
-         max_bytes: int,
-         window_seconds: float,
-      ) -> None:
-         self.limit_id = limit_id
-         self.url = url
-         self.max_bytes = max_bytes
-         self.window_seconds = window_seconds
-         self.user_body_sizes: dict[int, deque[tuple[float, int]]] = {}
-
-      def is_allowed(
-         self,
-         user_id: int,
-         request_url: str,
-         request_body: bytes,
-         request_timestamp: float,
-      ) -> bool:
-         if request_url != self.url:
-               return True
-
-         if user_id not in self.user_body_sizes:
-               self.user_body_sizes[user_id] = deque()
-
-         body_sizes = self.user_body_sizes[user_id]
-         cutoff_time = request_timestamp - self.window_seconds
-
-         while body_sizes and body_sizes[0][0] <= cutoff_time:
-               body_sizes.popleft()
-
-         current_total = sum(size for _, size in body_sizes)
-         return current_total + len(request_body) <= self.max_bytes
-
-      def record_request(
-         self,
-         user_id: int,
-         request_url: str,
-         request_body: bytes,
-         request_timestamp: float,
-      ) -> None:
-         if request_url != self.url:
-               return
-
-         self.user_body_sizes[user_id].append((request_timestamp, len(request_body)))
-
-
-   class RateLimiter:
-      def __init__(self) -> None:
-         self._limits: list[Limit] = [
-               RequestCountLimit(
-                  limit_id="v1_60s",
-                  urls=["/v1/completions.stream"],
-                  max_requests=3,
-                  window_seconds=60.0,
-               ),
-               RequestCountLimit(
-                  limit_id="v1_v2_3600s",
-                  urls=["/v1/completions.stream", "/v2/completions.stream"],
-                  max_requests=7,
-                  window_seconds=3600.0,
-               ),
-               BodySizeLimit(
-                  limit_id="v1_body_60s",
-                  url="/v1/completions.stream",
-                  max_bytes=100,
-                  window_seconds=60.0,
-               ),
-         ]
-
-      def allow_request(
-         self,
-         *,
-         user_id: int,
-         request_url: str,
-         request_body: bytes,
-         request_timestamp: float,
-      ) -> list[str]:
-         violated_limit_ids: list[str] = [
-               limit.limit_id
-               for limit in self._limits
-               if not limit.is_allowed(
-                  user_id=user_id,
-                  request_url=request_url,
-                  request_body=request_body,
-                  request_timestamp=request_timestamp,
-               )
-         ]
-
-         if violated_limit_ids:
-               return violated_limit_ids
-
-         for limit in self._limits:
-               limit.record_request(
-                  user_id=user_id,
-                  request_url=request_url,
-                  request_body=request_body,
-                  request_timestamp=request_timestamp,
-               )
-         return []
-
-   class BodySizeLimit:
-      def __init__(
-         self,
-         limit_id: str,
-         url: str,
-         max_bytes: int,
-         window_seconds: float,
-      ) -> None:
-         self.limit_id = limit_id
-         self.url = url
-         self.max_bytes = max_bytes
-         self.window_seconds = window_seconds
-         self.user_body_sizes: dict[int, deque[tuple[float, int]]] = {}
-
-      def is_allowed(
-         self,
-         user_id: int,
-         request_url: str,
-         request_body: bytes,
-         request_timestamp: float,
-      ) -> bool:
-         if request_url != self.url:
-               return True
-
-         if user_id not in self.user_body_sizes:
-               self.user_body_sizes[user_id] = deque()
-
-         body_sizes = self.user_body_sizes[user_id]
-         cutoff_time = request_timestamp - self.window_seconds
-
-         while body_sizes and body_sizes[0][0] <= cutoff_time:
-               body_sizes.popleft()
-
-         current_total = sum(size for _, size in body_sizes)
-         return current_total + len(request_body) <= self.max_bytes
-
-      def record_request(
-         self,
-         user_id: int,
-         request_url: str,
-         request_body: bytes,
-         request_timestamp: float,
-      ) -> None:
-         if request_url != self.url:
-               return
-
-         self.user_body_sizes[user_id].append((request_timestamp, len(request_body)))
-
-
-   class RateLimiter:
-      def __init__(self) -> None:
-         self._limits: list[Limit] = [
-               RequestCountLimit(
-                  limit_id="v1_60s",
-                  urls=["/v1/completions.stream"],
-                  max_requests=3,
-                  window_seconds=60.0,
-               ),
-               RequestCountLimit(
-                  limit_id="v1_v2_3600s",
-                  urls=["/v1/completions.stream", "/v2/completions.stream"],
-                  max_requests=7,
-                  window_seconds=3600.0,
-               ),
-               BodySizeLimit(
-                  limit_id="v1_body_60s",
-                  url="/v1/completions.stream",
-                  max_bytes=100,
-                  window_seconds=60.0,
-               ),
-         ]
-
-      def allow_request(
-         self,
-         *,
-         user_id: int,
-         request_url: str,
-         request_body: bytes,
-         request_timestamp: float,
-      ) -> list[str]:
-         violated_limit_ids: list[str] = [
-               limit.limit_id
-               for limit in self._limits
-               if not limit.is_allowed(
-                  user_id=user_id,
-                  request_url=request_url,
-                  request_body=request_body,
-                  request_timestamp=request_timestamp,
-               )
-         ]
-
-         if violated_limit_ids:
-               return violated_limit_ids
-
-         for limit in self._limits:
-               limit.record_request(
-                  user_id=user_id,
-                  request_url=request_url,
-                  request_body=request_body,
-                  request_timestamp=request_timestamp,
-               )
-         return []
+   """Python code."""
 
 
+   def hello_world() -> int:
+       """Return the answer."""
+       return 42
+
+
+   hello_world()
+
 .. code-block:: console
 
    $ pip install sphinx-notionbuilder