src/liquidcode/wsgi.py - master - liquid-g/liquid-code

src/liquidcode/wsgi.py
179 строк6 KB
User
0.3.0-RC - дописаны тесты и мелкие исправления
04 июл 2026, 13:07
04 июл 2026, 13:0740ad8b8
"""
WSGI-адаптер для LiquidCode.

Позволяет запускать приложение через серверы, поддерживающие WSGI (Gunicorn, uWSGI, Passenger).
"""

import json
import logging
from typing import Dict, List, Callable

from .application import Application
from .http import HttpRequest, HttpResponse
from .kernel import Kernel

logger = logging.getLogger(__name__)


class WSGIAdapter:
    """
    Адаптер для WSGI-сервера.

    Преобразует WSGI-окружение в HttpRequest, вызывает Kernel и возвращает WSGI-ответ.

    Пример:
        app = Application()
        wsgi_app = WSGIAdapter(app)

        # Для Gunicorn/Passenger:
        # gunicorn app:wsgi_app
    """

    def __init__(self, app: Application):
        """
        Args:
            app: Экземпляр Application (или Kernel).
        """
        self.app = app
        self.kernel: Kernel = app.kernel
        self._json = json  # кешируем импорт

    def __call__(
            self, environ: Dict[str, str], start_response: Callable
    ) -> List[bytes]:
        """
        WSGI-точка входа.

        Args:
            environ: WSGI-окружение.
            start_response: Функция для отправки статуса и заголовков.

        Returns:
            Итератор байтов для тела ответа.
        """
        # 1. Извлекаем данные запроса из environ
        method = environ.get("REQUEST_METHOD", "GET")

        # 2. Учитываем SCRIPT_NAME для корректной работы в подпапках
        script_name = environ.get("SCRIPT_NAME", "")
        path = environ.get("PATH_INFO", "/")

        # Если SCRIPT_NAME задан и путь начинается с него, убираем префикс
        if script_name and path.startswith(script_name):
            path = path[len(script_name):] or "/"

        query_string = environ.get("QUERY_STRING", "")
        if query_string:
            full_path = f"{path}?{query_string}"
        else:
            full_path = path

        headers = self._extract_headers(environ)
        body = self._extract_body(environ)

        # 3. Создаём HttpRequest
        request = HttpRequest(
            path=full_path,
            method=method,
            headers=headers,
            body=body,
        )

        # 4. Вызываем ядро с обработкой ошибок
        try:
            response = self.kernel.handle(request)
        except Exception as e:
            logger.exception("WSGI обработка запроса завершилась ошибкой")
            response = HttpResponse(
                content={"error": str(e)},
                status=500,
                headers={"Content-Type": "application/json"},
            )

        # 5. Преобразуем ответ в WSGI-формат
        status_code = response.get_status()
        status_text = self._get_status_text(status_code)
        response_headers = list(response.get_headers().items())

        content = response.get_content()

        # 6. Преобразуем содержимое в байты
        if isinstance(content, bytes):
            body_bytes = content
        else:
            body_bytes = self._json.dumps(content, ensure_ascii=False).encode("utf-8")
            # Добавляем Content-Type, если не указан
            if not any(k.lower() == "content-type" for k, _ in response_headers):
                response_headers.append(
                    ("Content-Type", "application/json; charset=utf-8")
                )

        # 7. Вызываем start_response
        start_response(f"{status_code} {status_text}", response_headers)

        # 8. Возвращаем тело (как список байтов)
        return [body_bytes]

    def _extract_headers(self, environ: Dict[str, str]) -> Dict[str, str]:
        """Извлекает заголовки из WSGI-окружения."""
        headers = {}
        for key, value in environ.items():
            if key.startswith("HTTP_"):
                # Преобразуем HTTP_HEADER_NAME в Header-Name
                header_name = key[5:].replace("_", "-").title()
                headers[header_name] = value
            elif key in ("CONTENT_TYPE", "CONTENT_LENGTH"):
                header_name = key.replace("_", "-").title()
                headers[header_name] = value
        return headers

    def _extract_body(self, environ: Dict[str, str]) -> bytes:
        """Извлекает тело запроса из WSGI-окружения."""
        try:
            content_length = int(environ.get("CONTENT_LENGTH", 0))
        except (ValueError, TypeError):
            content_length = 0

        if content_length <= 0:
            return b""

        try:
            return environ["wsgi.input"].read(content_length)
        except Exception:
            logger.exception("Ошибка чтения тела запроса")
            return b""

    def _get_status_text(self, status_code: int) -> str:
        """Возвращает стандартный текст для HTTP-статуса."""
        status_texts = {
            200: "OK",
            201: "Created",
            204: "No Content",
            400: "Bad Request",
            401: "Unauthorized",
            403: "Forbidden",
            404: "Not Found",
            405: "Method Not Allowed",
            500: "Internal Server Error",
        }
        return status_texts.get(status_code, "Unknown Status")


def make_wsgi_app(app: Application) -> WSGIAdapter:
    """
    Фабрика для создания WSGI-совместимого приложения из экземпляра Application.

    Это удобная обёртка для пользователей, которые не хотят создавать WSGIAdapter вручную.

    Args:
        app: Экземпляр Application.

    Returns:
        Экземпляр WSGIAdapter, готовый к использованию в WSGI-сервере.

    Пример:
        app = Application()
        # ... настройка
        wsgi_app = make_wsgi_app(app)
    """
    return WSGIAdapter(app)