src/liquidcode/wsgi.py - develop-0.4 - liquid-g/liquid-code

src/liquidcode/wsgi.py
212 строк8 KB
User
v0.4.17: рефакторинг статики и архитектура middleware
07 июл 2026, 12:05
07 июл 2026, 12:05706b43e
"""
WSGI-адаптер для LiquidCode.

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

Архитектура статики:
1. WSGIAdapter проверяет путь на префикс /static/
2. Если есть — пытается отдать файл из public/ (голая статика)
3. Если файл не найден — передаёт запрос в kernel.handle() (StaticFilesMiddleware обработает файл из static/)
"""

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

from .application import Application
from .http import HttpRequest, HttpResponse, try_serve_static_file
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  # кешируем импорт
        # Получаем настройки статики из Application
        self._static_dir = getattr(app, "_static_dir", "public")
        self._static_prefix = getattr(app, "_static_prefix", "/static")

    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. Попытка отдать статику напрямую (без обращения к ядру)
        static_success, static_response = try_serve_static_file(
            full_path, self._static_dir, self._static_prefix
        )
        if static_success:
            # Файл найден, отдаём его напрямую
            status_code = static_response.get_status()
            status_text = self._get_status_text(status_code)
            response_headers = list(static_response.get_headers().items())
            content = static_response.get_content()
            
            # Преобразуем содержимое в байты
            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")
                    )
            
            start_response(f"{status_code} {status_text}", response_headers)
            return [body_bytes]
        
        # 5. Вызываем ядро с обработкой ошибок
        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)