Implement ASGI application with fibonacci, factorial, and mean endpoints

hw1/app.py

@@ -1,19 +1,115 @@
 from typing import Any, Awaitable, Callable
+import json
+from http import HTTPStatus
+from router import Router, HTTPError, HTTPResponse, build_error_response
+from handlers import handle_factorial, handle_fibonacci, handle_mean
 
+router = Router()
+
+router.add_route('GET', '/factorial',  handle_factorial)
+router.add_route('GET', '/fibonacci/{n}',  handle_fibonacci)
+router.add_route('GET', '/mean',  handle_mean)
 
 async def application(
     scope: dict[str, Any],
     receive: Callable[[], Awaitable[dict[str, Any]]],
     send: Callable[[dict[str, Any]], Awaitable[None]],
 ):
     """
+    ASGI приложение с использованием шаблонного метода для роутинга
     Args:
         scope: Словарь с информацией о запросе
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
     """
-    # TODO: Ваша реализация здесь
+    if scope["type"] == "lifespan":
+        while True:
+            message = await receive()
+            if message["type"] == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+            elif message["type"] == "lifespan.shutdown":
+                await send({"type": "lifespan.shutdown.complete"})
+                return
+
+    elif scope["type"] == "http":
+        await handle_http_request(scope, receive, send)
+
+
+async def handle_http_request(
+        scope: dict[str, Any],
+        receive: Callable[[], Awaitable[dict[str, Any]]],
+        send: Callable[[dict[str, Any]], Awaitable[None]],
+):
+    """Обработка HTTP запросов"""
+    path = scope["path"]
+    method = scope["method"]
+    query_string = scope.get("query_string", b"").decode()
+
+    response = HTTPResponse(status=HTTPStatus.OK, body={ 'result': 'None'})
+    try:
+        body = await parse_request_body(scope, receive)
+
+        handler, context = await router.route(method, path, query_string, body)
+
+        response: HTTPResponse = await handler(context)
+
+    except HTTPError as e:
+        response = build_error_response(e.status, e.error, e.message)
+    except Exception as e:
+        response = build_error_response(HTTPStatus.INTERNAL_SERVER_ERROR, 'Internal Server Error', str(e))
+    finally:
+        await send_response(send, response)
+
+
+async def send_response(send: Callable, response: HTTPResponse):
+    """Отправка ответов"""
+    response_body = json.dumps(response['body']).encode("utf-8")
+
+    await send({
+        'type': 'http.response.start',
+        'status': response['status'],
+        'headers': [
+            (b'content-type', b'application/json'),
+            (b'content-length', str(len(response_body)).encode())
+        ],
+    })
+
+    await send({
+        'type': 'http.response.body',
+        'body': response_body,
+    })
+
+async def parse_request_body(scope: dict[str, Any], receive: Callable) -> Any:
+    """Загрузка тела запроса"""
+    headers = dict(scope.get("headers", []))
+    content_length = headers.get(b"content-length")
+
+    if content_length:
+        try:
+            content_length = int(content_length.decode())
+        except (ValueError, AttributeError):
+            content_length = 0
+    else:
+        content_length = 0
+
+    if content_length <= 0:
+        return None
+
+    body = b""
+    more_body = True
+    while more_body:
+        message = await receive()
+        body += message.get("body", b"")
+        more_body = message.get("more_body", False)
+
+    if not body:
+        return None
+
+    try:
+        return json.loads(body.decode("utf-8"))
+    except json.JSONDecodeError:
+        raise HTTPError(HTTPStatus.UNPROCESSABLE_ENTITY, "Unprocessable Entity", "Invalid JSON format")
 
 if __name__ == "__main__":
     import uvicorn
-    uvicorn.run("app:application", host="0.0.0.0", port=8000, reload=True)
+    uvicorn.run("app:application", host="0.0.0.0", port=8000, reload=True)

hw1/handlers.py

@@ -0,0 +1,78 @@
+from structures import HTTPError, HTTPResponse
+from http import HTTPStatus
+import math
+
+async def handle_factorial(context: dict) -> HTTPResponse:
+    """Обработка факториала"""
+    query_params = context['query_params']
+
+    if 'n' not in query_params:
+        raise HTTPError(HTTPStatus.UNPROCESSABLE_ENTITY, "Unprocessable Entity", "Parameter 'n' is required")
+
+    try:
+        n = int(query_params['n'])
+    except ValueError:
+        raise HTTPError(HTTPStatus.UNPROCESSABLE_ENTITY, "Unprocessable Entity", "Parameter 'n' must be an integer")
+
+    if n < 0:
+        raise HTTPError(HTTPStatus.BAD_REQUEST, "Bad Request", "Parameter 'n' must be non-negative")
+
+    try:
+        result = math.factorial(n)
+        return {
+            'status': HTTPStatus.OK,
+            'body': {'result': result}
+        }
+    except OverflowError:
+        raise HTTPError(HTTPStatus.BAD_REQUEST, "Bad Request", "Parameter 'n' is too large")
+
+
+async def handle_fibonacci(context: dict) -> HTTPResponse:
+    """Обработка чисел Фибоначчи"""
+    path_params = context['path_params']
+
+    try:
+        n = int(path_params['n'])
+    except (KeyError, ValueError):
+        raise HTTPError(HTTPStatus.UNPROCESSABLE_ENTITY, "Unprocessable Entity", "Invalid number parameter")
+
+    if n < 0:
+        raise HTTPError(HTTPStatus.BAD_REQUEST, "Bad Request", "Number must be non-negative")
+
+    if n in [0, 1]:
+        result = n
+    else:
+        a, b = 0, 1
+        for _ in range(2, n + 1):
+            a, b = b, a + b
+        result = b
+
+    return {
+        'status': HTTPStatus.OK,
+        'body': {'result': result}
+    }
+
+
+async def handle_mean(context: dict) -> HTTPResponse:
+    """Обработка среднего значения"""
+    body_data = context['body']
+
+    if body_data is None:
+        raise HTTPError(HTTPStatus.UNPROCESSABLE_ENTITY, "Unprocessable Entity", "JSON body is required")
+
+    if not isinstance(body_data, list):
+        raise HTTPError(HTTPStatus.UNPROCESSABLE_ENTITY, "Unprocessable Entity", "Expected a list of numbers")
+
+    if not body_data:
+        raise HTTPError(HTTPStatus.BAD_REQUEST, "Bad Request", "At least one number is required")
+
+    try:
+        numbers = [float(num) for num in body_data]
+    except (TypeError, ValueError):
+        raise HTTPError(HTTPStatus.UNPROCESSABLE_ENTITY, "Unprocessable Entity", "All values must be valid numbers")
+
+    mean = sum(numbers) / len(numbers)
+    return {
+        'status': HTTPStatus.OK,
+        'body': {'result': mean}
+    }

hw1/router.py

@@ -0,0 +1,77 @@
+from typing import Any, Callable, Optional, Dict, TypedDict
+import re
+from http import HTTPStatus
+from structures import HTTPResponse, HTTPError
+
+def build_error_response(status: int, error: str, message: str) -> HTTPResponse:
+    response: HTTPResponse = {
+        'status': status,
+        'body': {
+            'error': error,
+            'message': message
+        }
+    }
+    return response
+
+class Router:
+
+    def __init__(self):
+        self.routes = []
+        self.handlers: Dict[str, Callable] = {}
+
+    def add_route(self, method: str, path: str, handler: Callable):
+        """Добавляет маршрут"""
+        pattern = self._path_to_pattern(path)
+        self.routes.append({
+            'method': method.upper(),
+            'pattern': pattern,
+            'handler': handler
+        })
+
+    @staticmethod
+    def _path_to_pattern(path: str) -> re.Pattern:
+        """Конвертирует путь с параметрами в regex паттерн"""
+        pattern = re.sub(r'\{(\w+)}', r'(?P<\1>[^/]+)', path)
+        return re.compile(f'^{pattern}$')
+
+    async def route(self, method: str, path: str, query_string: str, body: Optional[Any]) -> tuple[Callable, dict]:
+        method = method.upper()
+
+        route_match = self._match_route(method, path)
+        if not route_match:
+            raise HTTPError(HTTPStatus.NOT_FOUND, "Not Found", "Endpoint not found")
+
+        handler, path_params = route_match
+
+        query_params = self._parse_query_string(query_string)
+
+        request_context = {
+            'path_params': path_params,
+            'query_params': query_params,
+            'body': body
+        }
+
+        return handler, request_context
+
+    def _match_route(self, method: str, path: str) -> Optional[tuple[Callable, dict]]:
+        """Ищет подходящий маршрут и возвращает handler"""
+        for route in self.routes:
+            if route['method'] != method:
+                continue
+
+            match = route['pattern'].match(path)
+            if match:
+                return route['handler'], match.groupdict()
+
+        return None
+
+    @staticmethod
+    def _parse_query_string(query_string: str) -> dict:
+        """Парсит query string в словарь"""
+        params = {}
+        if query_string:
+            for pair in query_string.split('&'):
+                if '=' in pair:
+                    key, value = pair.split('=', 1)
+                    params[key] = value
+        return params

hw1/structures.py

@@ -0,0 +1,15 @@
+from typing import TypedDict
+
+class HTTPResponse(TypedDict):
+    """Типизированный ответ обработчика"""
+    status: int
+    body: dict
+
+class HTTPError(Exception):
+    """Кастомное исключение для HTTP ошибок"""
+
+    def __init__(self, status: int, error: str, message: str):
+        self.status = status
+        self.error = error
+        self.message = message
+        super().__init__(f"{error}: {message}")

hw2/hw/shop_api/chat/websocket.py

@@ -0,0 +1,93 @@
+import json
+import uuid
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect
+from typing import Dict, List
+import asyncio
+
+
+class ConnectionManager:
+    def __init__(self):
+        self.active_connections: Dict[str, List[WebSocket]] = {}
+        self.usernames: Dict[WebSocket, str] = {}
+
+    async def connect(self, websocket: WebSocket, chat_name: str):
+        await websocket.accept()
+
+        if chat_name not in self.active_connections:
+            self.active_connections[chat_name] = []
+
+        self.active_connections[chat_name].append(websocket)
+
+        username = f"User_{uuid.uuid4().hex[:8]}"
+        self.usernames[websocket] = username
+
+        await self.broadcast_message(
+            chat_name,
+            f"🟢 {username} :: присоединился к чату",
+            exclude_websocket=websocket
+        )
+
+        await websocket.send_text(f"🤖 Добро пожаловать в чат '{chat_name}'! Ваше имя: {username}")
+
+    def disconnect(self, websocket: WebSocket, chat_name: str):
+        if chat_name in self.active_connections:
+            if websocket in self.active_connections[chat_name]:
+                self.active_connections[chat_name].remove(websocket)
+
+            if not self.active_connections[chat_name]:
+                del self.active_connections[chat_name]
+
+        if websocket in self.usernames:
+            username = self.usernames[websocket]
+            del self.usernames[websocket]
+
+            if chat_name in self.active_connections:
+                self.broadcast_message_sync(
+                    chat_name,
+                    f"🔴 {username} :: покинул чат",
+                    exclude_websocket=websocket
+                )
+
+    async def broadcast_message(self, chat_name: str, message: str, exclude_websocket: WebSocket = None):
+        if chat_name in self.active_connections:
+            for connection in self.active_connections[chat_name]:
+                if connection != exclude_websocket:
+                    try:
+                        await connection.send_text(message)
+                    except:
+                        self.disconnect(connection, chat_name)
+
+    def broadcast_message_sync(self, chat_name: str, message: str, exclude_websocket: WebSocket = None):
+        """Синхронная версия для использования в disconnect"""
+        if chat_name in self.active_connections:
+            disconnected = []
+            for connection in self.active_connections[chat_name]:
+                if connection != exclude_websocket:
+                    try:
+                        asyncio.create_task(connection.send_text(message))
+                    except:
+                        disconnected.append(connection)
+
+            for connection in disconnected:
+                self.disconnect(connection, chat_name)
+
+    async def handle_message(self, websocket: WebSocket, chat_name: str, message: str):
+        username = self.usernames.get(websocket, "Unknown")
+        formatted_message = f"{username} :: {message}"
+        await self.broadcast_message(chat_name, formatted_message)
+
+
+manager = ConnectionManager()
+chat_router = APIRouter()
+
+
+@chat_router.websocket("/chat/{chat_name}")
+async def websocket_endpoint(websocket: WebSocket, chat_name: str):
+    await manager.connect(websocket, chat_name)
+
+    try:
+        while True:
+            data = await websocket.receive_text()
+            await manager.handle_message(websocket, chat_name, data)
+    except WebSocketDisconnect:
+        manager.disconnect(websocket, chat_name)