How to create your own web framework in Python
Good day, habr! In this article, we will create our own web framework in Python using gunicorn. It will be lightweight and have basic functionality. We will create request handlers (views), simple and parameterized routing, Middleware, i18n and l10n, Request/Response, HTML templa
Editor's Context
This article is an English adaptation with additional editorial framing for an international audience.
- Terminology and structure were localized for clarity.
- Examples were rewritten for practical readability.
- Technical claims were preserved with source attribution.
Source: original publication
Good day, habr! In this article, we will create our own web framework in Python using gunicorn.
It will be lightweight and have basic functionality. We will create request handlers (views), simple and parameterized routing, Middleware, i18n and l10n, Request/Response, HTML template processing and documentation generation.
In this article we will build the most important parts of the framework, study how WSGI works and create web applications. And it will also be easier for us in the future to understand the logic of other frameworks: flask, django.
Some of you may say that we are reinventing the wheel. And I’ll say in response: can you right now, without any hints, just from memory, draw a bicycle without mistakes?
The most important parts of web frameworks are:
Routing handlers:
Simple:
/indexParameterized:
/article/{article_id}
Request handlers (views).
Middleware
Request/Response
i18n/l10n
Configuration
Basic requirement: the web framework must be supported by a fast, lightweight and efficient server (eg gunicorn). Python has a WSGI guide for this.
Our web framework will be called pyEchoNext. Repository link.
❯ Python web server design
ЗАПРОС CLIENT <--------------> [HTTP (80) или HTTPS (443)] Сервер ОТВЕТ > Приложение с логикой > Преобразование данных для python-приложения <-- Зона интересов веб-фреймвока (обеспечение работы gunicorn с ним) > Gunicorn > Преобразованные данные СЕРВЕР -> NGINX > Маршрутизация данных
When developing a web application in python, we encounter the following problems:
Many frameworks (ex. django) do not know how to route response requests.
Applications are insecure and may be susceptible to DDoS (Distributed Denial of Service) attacks.
There is no load balancing between multiple servers.
NGINX solves the problem of load balancing, but it cannot run and communicate with Python applications.
This is why there is a need to use a WSGI server (Web Server Gateway Interface) and a proxy server (such as NGINX).
❯ WSGI
Python currently boasts a wide range of web application frameworks such as Zope, Quixote, Webware, SkunkWeb, PSO and Twisted Web, just to name a few. This wide variety of options can be a challenge for new Python users, as typically their choice of web framework will limit their choice of web servers to use, and vice versa.
In contrast, although Java has as many web application frameworks available, Java's "servlet" API allows applications written with any Java web application framework to run on any web server that supports the servlet API.
The availability and widespread use of such an API in web servers for Python—whether those servers are written in Python (e.g., Medusa), built in Python (e.g., mod_python), or call Python through a gateway protocol (e.g., CGI, FastCGI, etc.)—will decouple the choice of framework from the choice of web server, allowing users to choose the pairing that suits them, while freeing up the framework and server developers to focus on their preferred area specializations.
Thus, this PEP offers a simple and universal interface between web servers and web applications or frameworks: the Python Web Server Gateway Interface (WSGI).
But the mere existence of the WSGI specification does nothing to address the current state of Python web application servers and frameworks. Authors and maintainers of servers and frameworks must actually implement WSGI for it to have any effect.
However, since no existing server or framework supports WSGI, an author who implements WSGI support will not receive immediate rewards. Thus, WSGI must be easy to implement so that the author's initial investment in the interface can be fairly low.
Thus, ease of implementation on both the server side and the interface framework side is absolutely critical to the usefulness of a WSGI interface and is therefore a primary criterion for any design decisions.
However, it should be noted that ease of implementation for a framework author is not the same as ease of use for a web application author. WSGI provides a completely "no frills" interface for the framework author, because bells and whistles like response objects and cookie handling would simply prevent existing frameworks from solving these problems. Again, the goal of WSGI is to facilitate simple interoperability between existing servers and applications or frameworks, not to create a new web framework.
It should also be noted that this target does not allow WSGI to require anything that is not already available in deployed versions of Python. Therefore, new standard library modules are not proposed or required by this specification, and nothing in WSGI requires a Python version greater than 2.2.2. (However, it would be nice if future versions of Python included support for this interface in the web servers provided by the standard library).
In addition to being easy to implement for existing and future frameworks and servers, it should also be easy to create request preprocessors, response postprocessors, and other WSGI-based "middleware" components that look like an application to its containing server while acting as a server to its contained applications. If middleware can be both simple and robust, and WSGI is widely available in servers and frameworks, this allows for the possibility of an entirely new type of Python web application framework: consisting of loosely coupled WSGI middleware components. Indeed, existing framework authors may even choose to refactor their frameworks' existing services so that they are exposed in a way that becomes more like the libraries used with WSGI and less like monolithic frameworks. This would then allow application developers to select "best-of-breed" components for a specific functionality, rather than committing to all the pros and cons of a single framework.
Of course, as of this writing, that day is undoubtedly quite far away. At the same time, this is a sufficient short-term goal for WSGI to enable the use of any framework with any server.
Finally, it should be mentioned that the current version of WSGI does not prescribe any specific mechanism for "deploying" an application for use with a web server or server gateway. Currently, this is necessarily determined by the server or gateway implementation. Once enough servers and frameworks have implemented WSGI to provide hands-on experience with various deployment requirements, it may make sense to create another PEP describing
❯ Goals of pyEchoNext
pyEchoNext is a universal tool with the ability to create a monolithic web application, or vice versa, a modular web application. Django was too big and clumsy for us, flask or fastapi were too small. Therefore, we decided to take some features from django and flask/fastapi, combine them and make it all symbiotic. So that you can make a large monolithic project or a small service. And to turn a small service into a large application or vice versa, a minimum of effort was required.
Our goals were also to make all this as clear as possible, developer-friendly, and add the ability to integrate third-party libraries.
As a result, the main characteristics of the project are as follows:
Goal: Create a universal multi-faceted web framework in python.
Tasks:
Find the good and bad sides of Flask, FastAPI
Find the good and bad sides of Django
Compare the capabilities of existing frameworks
Selecting the best features
Symbiosis of features into one whole
Build project code according to SOLID and OOP principles, easily extensible, scalable and complementary.
Make the code fast and productive, give freedom to the user and developer
Problem: at the moment there are very few universal frameworks that allow you to create both a large monolithic application and a fast small service.
Relevance: the web sphere is very popular at the moment, the ability to work with web frameworks, abstractions, and know the structure of sites will help everyone.
❯ How running a web application through gunicorn works
Install gunicorn And pysocks for subsequent actions.
So create an app.py file:
from socks import method def app(environ: dict, start_response: method): response_body = b'Hello, Habr!' status = "200 OK" start_response(status, headers=[]) return iter([response_body])
And then run gunicorn:
gunicorn app:app # gunicorn <файл>:<callable-класс или функция точки входа>
The entry point receives two parameters - environ and start_response. environ contains all the information about the web environment, such as user-agent, path, method, GET and POST parameters and others. The second parameter is start_response, the start response that sends the expected response.
But a better practice would be to create a callable class:
class App: def __call__(self, environ: dict, start_response: method): response_body = b'Hello, Habr!' status = "200 OK" start_response(status, headers=[]) return iter([response_body]) app = App()
Magic method __call__ makes objects of our class callable.
And now you can absolutely run the application:
gunicorn app:app # gunicorn <файл>:<callable-класс или функция точки входа>
But let's now gradually fill out our project, filling it with various modules. Let's start by creating a project through poetry.
❯ Creating a project
Poetry is a tool for managing dependencies and building packages in Python. Poetry also makes it very easy to publish your library on PyPi!
Poetry provides the complete set of tools you need for deterministic project management in Python. Including building packages, supporting different versions of the language, testing and deploying projects.
It all started when Poetry creator Sébastien Eustace wanted a single tool to manage projects from start to finish, reliable and intuitive, that could also be used within the community. A dependency manager alone was not enough to manage the running of tests, the deployment process, and the entire co-dependent environment. This functionality is beyond the capabilities of conventional package managers such as Pip or Conda. This is how Python Poetry was born.
You can install poetry via pipx: pipx install poetry and via pip: pip install poetry --break-system-requirements. This will install poetry globally across the entire system.
So let's create a project using poetry and install the dependencies:
poetry new <имя_проекта> cd <имя_проекта> poetry shell poetry add ruff loguru pysocks fire python-dotenv jinja2 parse gunicorn configparser
❯ Project architecture
I ended up with the following project architecture:
pyechonext/ ├── apidoc_ui │ ├── api_documentation.py │ └── __init__.py ├── app.py ├── config.py ├── docsgen │ ├── document.py │ ├── __init__.py │ └── projgen.py ├── i18n_l10n │ ├── i18n.py │ └── l10n.py ├── __init__.py ├── logging │ ├── __init__.py │ └── logger.py ├── __main__.py ├── middleware.py ├── request.py ├── response.py ├── template_engine │ ├── builtin.py │ ├── __init__.py │ └── jinja.py ├── urls.py ├── utils │ ├── exceptions.py │ └── __init__.py └── views.py
The apidoc_ui directory is the generation of OpenAPI project documentation.
The docsgen directory is where project documentation is generated.
Directory i18n_l10n - internationalization and localization.
The logging directory is logging.
The template_engine directory — html template engines.
utils directory - utilities.
The app.py file is the application.
File config.py - configuration and loading of settings.
File
__main__.py— main module, to be launched viapython3 -m pyechonext.The middleware.py file is the middleware.
The request.py file is the request class.
The response.py file is the response class.
File urls.py - URLs (handlers).
Views.py file - request handlers.
❯ We implement custom exceptions
Exceptions are an integral part of the web framework. I decided to implement several parent classes:
pyEchoNextException - base exception.
WebError - web error (inherits from pyEchoNextException). It differs in that it has an HTTP error code.
Therefore, there are the following exceptions that inherit from pyEchoNextException:
InternationalizationNotFound—The internationalization file was not found.
LocalizationNotFound - localization file not found.
TemplateNotFileError - the template is not a file.
RoutePathExistsError - The route path already exists.
And the following exceptions are descendants of WebError:
URLNotFound - URL not found (404).
MethodNotAllow - the method is not allowed (405).
TeapotError - the server is a teapot (418).
Source code for custom exceptions
from loguru import logger class pyEchoNextException(Exception): """ Exception for signaling pyechonext errors. """ def __init__(self, *args): """ Constructs a new instance. :param args: The arguments :type args: list """ if args: self.message = args[0] else: self.message = None def get_explanation(self) -> str: """ Gets the explanation. :returns: The explanation. :rtype: str """ return f"Message: {self.message if self.message else 'missing'}" def __str__(self): """ Returns a string representation of the object. :returns: String representation of the object. :rtype: str """ logger.error(f"{self.__class__.__name__}: {self.get_explanation()}") return f"pyEchoNextException has been raised. {self.get_explanation()}" class WebError(pyEchoNextException): code = 400 def get_explanation(self) -> str: """ Gets the explanation. :returns: The explanation. :rtype: str """ return ( f"Code: {self.code}. Message: {self.message if self.message else 'missing'}" ) def __str__(self): """ Returns a string representation of the object. :returns: String representation of the object. :rtype: str """ logger.error(f"{self.__class__.__name__}: {self.get_explanation()}") return f"WebError has been raised. {self.get_explanation()}" class InternationalizationNotFound(pyEchoNextException): def __str__(self): """ Returns a string representation of the object. :returns: String representation of the object. :rtype: str """ logger.error(f"{self.__class__.__name__}: {self.get_explanation()}") return f"InternationalizationNotFound has been raised. {self.get_explanation()}" class LocalizationNotFound(pyEchoNextException): def __str__(self): """ Returns a string representation of the object. :returns: String representation of the object. :rtype: str """ logger.error(f"{self.__class__.__name__}: {self.get_explanation()}") return f"LocalizationNotFound has been raised. {self.get_explanation()}" class TemplateNotFileError(pyEchoNextException): def __str__(self): """ Returns a string representation of the object. :returns: String representation of the object. :rtype: str """ logger.error(f"{self.__class__.__name__}: {self.get_explanation()}") return f"TemplateNotFileError has been raised. {self.get_explanation()}" class RoutePathExistsError(pyEchoNextException): def __str__(self): """ Returns a string representation of the object. :returns: String representation of the object. :rtype: str """ logger.error(f"{self.__class__.__name__}: {self.get_explanation()}") return f"RoutePathExistsError has been raised. {self.get_explanation()}" class URLNotFound(WebError): code = 404 def __str__(self): """ Returns a string representation of the object. :returns: String representation of the object. :rtype: str """ logger.error(f"{self.__class__.__name__}: {self.get_explanation()}") return f"URLNotFound has been raised. {self.get_explanation()}" class MethodNotAllow(WebError): code = 405 def __str__(self): """ Returns a string representation of the object. :returns: String representation of the object. :rtype: str """ logger.error(f"{self.__class__.__name__}: {self.get_explanation()}") return f"MethodNotAllow has been raised. {self.get_explanation()}" class TeapotError(WebError): code = 418 def __str__(self): """ Returns a string representation of the object. :returns: String representation of the object. :rtype: str """ logger.error(f"{self.__class__.__name__}: {self.get_explanation()}") return f"The server refuses to make coffee because he is a teapot. {self.get_explanation()}"
❯ We implement internationalization and localization
i18n is an abbreviation for internationalization process.
l10n - localization, that is, the process of taking into account culture and the rules for writing dates, monetary amounts, and numbers.
Internationalization is the process of developing an application in which its code is independent of any linguistic and cultural characteristics of the region or country. As a result, the application becomes flexible and can easily adapt to different language and cultural settings.
Internationalization implementation usually begins early in the project to prepare the product for future localization. During this process, they determine what will change for future locales (for example, text, images) and export this data to external files.
The 18 in i18n stands for the number of letters between the first letter i and the last letter n in the word "internationalization".
In our project there will be two classes in the i18n.py file - the abstract i18nInterface. It has two abstract methods - load_locale and get_string.
The JSONi18nLoader class inherits from it; it loads internationalization from a json file. But it has a default localization:
DEFAULT_LOCALE = { "title": "pyEchoNext Example Website", "description": "This web application is an example of the pyEchonext web framework.", }
json file should have the following name: <локаль>.json. For example, for RU_RU: RU_RU.json. And it looks like this:
{ "i18n": { "title": "pyEchoNext Веб-приложение с локалью", "example one": "пример один" }, "l10n": { "date_format": "%Y-%m-%d", "time_format": "%H:%M", "date_time_fromat": "%Y-%m-%d %H:%M", "thousands_separator": ",", "decimal_separator": ".", "currency_symbol": "$", "currency_format": "{symbol}{amount}" } }
As you can see, one json file contains both i18n and l10n.
The l10n.py file has the same structure - an abstract LocalizationInterface class with abstract methods load_locale, format_date, format_number, format_currency, get_current_settings and update_settings.
And absolutely there is also a JSONLocalizationLoader class (inherited from the interface). It already has the following default parameters:
DEFAULT_LOCALE = { "date_format": "%Y-%m-%d", "time_format": "%H:%M", "date_time_fromat": "%Y-%m-%d %H:%M", "thousands_separator": ",", "decimal_separator": ".", "currency_symbol": "$", "currency_format": "{symbol}{amount}", }
These parameters must be required in the locale file, otherwise the non-existent parameter will be replaced with the default one.
Source code i18n.py
import json import os from abc import ABC, abstractmethod from typing import Dict from loguru import logger from pyechonext.utils.exceptions import InternationalizationNotFound class i18nInterface(ABC): """ This class describes a locale interface. """ @abstractmethod def get_string(self, key: str) -> str: """ Gets the string. :param key: The key :type key: str :returns: The string. :rtype: str """ raise NotImplementedError @abstractmethod def load_locale(self, locale: str, directory: str) -> Dict[str, str]: """ Loads a locale. :param locale: The locale :type locale: str :param directory: The directory :type directory: str :returns: locale translations :rtype: Dict[str, str] """ raise NotImplementedError class JSONi18nLoader(i18nInterface): """ This class describes a json locale loader. """ DEFAULT_LOCALE = { "title": "pyEchoNext Example Website", "description": "This web application is an example of the pyEchonext web framework.", } def __init__(self, locale: str, directory: str): """ Constructs a new instance. :param locale: The locale :type locale: str :param directory: The directory :type directory: str """ self.locale: str = locale self.directory: str = directory self.translations: Dict[str, str] = self.load_locale( self.locale, self.directory ) def load_locale(self, locale: str, directory: str) -> Dict[str, str]: """ Loads a locale. :param locale: The locale :type locale: str :param directory: The directory :type directory: str :returns: locale dictionary :rtype: Dict[str, str] """ if self.locale == "DEFAULT": return self.DEFAULT_LOCALE file_path = os.path.join(self.directory, f"{self.locale}.json") try: logger.info(f"Load locale: {file_path} [{self.locale}]") with open(file_path, "r", encoding="utf-8") as file: i18n = json.load(file).get("i18n", None) if i18n is None: return json.load(file) else: return i18n except FileNotFoundError: raise InternationalizationNotFound( f"[i18n] i18n file at {file_path} not found" ) def get_string(self, key: str, **kwargs) -> str: """ Gets the string. :param key: The key :type key: str :param kwargs: The keywords arguments :type kwargs: dictionary :returns: The string. :rtype: str """ result = "" for word in key.split(" "): result += f"{self.translations.get(word, word)} " if kwargs: for name, value in kwargs.items(): result = result.replace(f'{f"%{{{name}}}"}', value) return result.strip() class LanguageManager: """ This class describes a language manager. """ def __init__(self, loader: i18nInterface): """ Constructs a new instance. :param loader: The loader :type loader: i18nInterface """ self.loader = loader def translate(self, key: str) -> str: """ Translate :param key: The key :type key: str :returns: translated string :rtype: str """ return self.loader.get_string(key)
Source code l10n.py
import json import os from datetime import datetime from abc import ABC, abstractmethod from typing import Dict, Any, Optional from loguru import logger from pyechonext.utils.exceptions import LocalizationNotFound class LocalizationInterface(ABC): """ This class describes a locale interface. """ @abstractmethod def load_locale(self, locale: str, directory: str) -> Dict[str, str]: """ Loads a locale. :param locale: The locale :type locale: str :param directory: The directory :type directory: str :returns: locale translations :rtype: Dict[str, str] """ raise NotImplementedError @abstractmethod def format_date(self, date: datetime, date_format: Optional[str] = None) -> str: """ Format date :param date: The date :type date: datetime :returns: formatted date :rtype: str """ raise NotImplementedError @abstractmethod def format_number(self, number: float, decimal_places: int = 2) -> str: """ Format number :param number: The number :type number: float :param decimal_places: The decimal places :type decimal_places: int :returns: formatted number :rtype: str """ raise NotImplementedError @abstractmethod def format_currency(self, amount: float) -> str: """ Format currency :param amount: The amount :type amount: float :returns: formatted currency :rtype: str """ raise NotImplementedError @abstractmethod def get_current_settings(self) -> Dict[str, Any]: """ Gets the current settings. :returns: The current settings. :rtype: Dict[str, Any] """ raise NotImplementedError @abstractmethod def update_settings(self, settings: Dict[str, Any]): """ Update settings :param settings: The settings :type settings: Dict[str, Any] """ raise NotImplementedError class JSONLocalizationLoader(LocalizationInterface): """ This class describes a json localization loader. """ DEFAULT_LOCALE = { "date_format": "%Y-%m-%d", "time_format": "%H:%M", "date_time_fromat": "%Y-%m-%d %H:%M", "thousands_separator": ",", "decimal_separator": ".", "currency_symbol": "$", "currency_format": "{symbol}{amount}", } def __init__( self, locale: str, directory: str, custom_settings: Optional[Dict[str, Any]] = None, ): """ Constructs a new instance. :param locale: The locale :type locale: str :param directory: The directory :type directory: str :param custom_settings: The custom settings :type custom_settings: Optional[Dict[str, Any]] """ self.locale: str = locale self.directory: str = directory self.locale_settings: Dict[str, Any] = self.load_locale(locale, directory) if custom_settings: self.update_settings(custom_settings) def load_locale(self, locale: str, directory: str) -> Dict[str, str]: """ Loads a locale. :param locale: The locale :type locale: str :param directory: The directory :type directory: str :returns: locale dictionary :rtype: Dict[str, str] """ if self.locale == "DEFAULT": return self.DEFAULT_LOCALE file_path = os.path.join(self.directory, f"{self.locale}.json") try: logger.info(f"Load locale: {file_path} [{self.locale}]") with open(file_path, "r", encoding="utf-8") as file: l10n = json.load(file).get("l10n", None) if l10n is None: return json.load(file) else: return l10n except FileNotFoundError: raise LocalizationNotFound(f"[l10n] l10n file at {file_path} not found") def format_date(self, date: datetime, date_format: Optional[str] = None) -> str: """ Format date :param date: The date :type date: datetime :param date_format: The date format :type date_format: Optional[str] :returns: formatted date :rtype: str """ date_time_fromat = ( self.locale_settings.get( "date_time_fromat", self.DEFAULT_LOCALE["date_time_fromat"] ) if date_format is None else date_format ) formatted_date = date_time_fromat.strftime(date_time_fromat) return formatted_date def format_number(self, number: float, decimal_places: int = 2) -> str: """ Format number :param number: The number :type number: float :param decimal_places: The decimal places :type decimal_places: int :returns: formatted number :rtype: str """ thousands_separator = self.locale_settings.get( "thousands_separator", self.DEFAULT_LOCALE["thousands_separator"] ) decimal_separator = self.locale_settings.get( "decimal_separator", self.DEFAULT_LOCALE["decimal_separator"] ) formatted_number = ( f"{number:,.{decimal_places}f}".replace(",", "TEMP") .replace(".", decimal_separator) .replace("TEMP", thousands_separator) ) return formatted_number def format_currency(self, amount: float) -> str: """ Format currency :param amount: The amount :type amount: float :returns: formatted currency :rtype: str """ currency_symbol = self.locale_settings.get( "currency_symbol", self.DEFAULT_LOCALE["currency_symbol"] ) currency_format = self.locale_settings.get( "currency_format", self.DEFAULT_LOCALE["currency_format"] ) return currency_format.format( symbol=currency_symbol, amount=self.format_number(amount) ) def update_settings(self, settings: Dict[str, Any]): """ Update settings :param settings: The settings :type settings: Dict[str, Any] :raises ValueError: setting is not recognized """ for key, value in settings.items(): if key in self.locale_settings: self.locale_settings[key] = value elif key in self.DEFAULT_LOCALE: self.DEFAULT_LOCALE[key] = value else: raise ValueError(f'[l10n] Setting "{key}" is not recognized.') def get_current_settings(self) -> Dict[str, Any]: """ Gets the current settings. :returns: The current settings. :rtype: Dict[str, Any] """ return { "locale": self.locale, "directory": self.directory, **self.locale_settings, **self.DEFAULT_LOCALE, }
❯ We implement logging
We have still imported from logger import loguru, but I didn't explain what it is. Loguru is a more convenient alternative wrapper around logging. To configure it, create a file logging/logger.py:
import logging from typing import Union, List from loguru import logger class InterceptHandler(logging.Handler): """ This class describes an intercept handler. """ def emit(self, record) -> None: """ Get corresponding Loguru level if it exists :param record: The record :type record: record :returns: None :rtype: None """ try: level = logger.level(record.levelname).name except ValueError: level = record.levelno frame, depth = logging.currentframe(), 2 while frame.f_code.co_filename == logging.__file__: frame = frame.f_back depth += 1 logger.opt(depth=depth, exception=record.exc_info).log( level, record.getMessage() ) def setup_logger(level: Union[str, int] = "DEBUG", ignored: List[str] = "") -> None: """ Setup logger :param level: The level :type level: str :param ignored: The ignored :type ignored: List[str] """ logging.basicConfig( handlers=[InterceptHandler()], level=logging.getLevelName(level) ) for ignore in ignored: logger.disable(ignore) logger.add("pyechonext.log") logger.info("Logging is successfully configured")
In the code above we assign a log file and configure it.
❯ Generation of project documentation
A little hello from my last article about managing project documentation using python.
I will not describe all the code, you can see and integrate it into your project from my article.
But I added one file - docsgen/projgen.py, it is responsible for generating:
from typing import Callable, Any from pyechonext.app import EchoNext from pyechonext.docsgen.document import ( InitiationSection, DocumentFolder, ProjectManager, ProjectTemplate, RoutesSubsection, DocumentSection, ) class ProjDocumentation: """ This class describes an api documentation. """ def __init__(self, echonext_app: EchoNext): """ Constructs a new instance. :param echonext_app: The echonext application :type echonext_app: EchoNext """ self.app = echonext_app self.app_name = echonext_app.app_name self.pages = {} def generate_documentation(self): """ Generate documentation """ section = self._generate_introduction() self._generate_subsections(section) folder = DocumentFolder( "api", f"{self.app_name}/docs", [ section, ], ) project_manager = ProjectManager( f"{self.app_name}", "Project Web Application", "Project application based on pyEchoNext web-framework", f"{self.app_name}", f"{self.app_name}", f"{self.app_name}", ProjectTemplate.BASE, [folder], [section], ) project_manager.process_project() def _generate_introduction(self) -> InitiationSection: """ Generate introduction :returns: The initiation section. :rtype: InitiationSection """ section = InitiationSection( f"Project {self.app_name}", f"Project Documentation for {self.app_name}", {"Routes": ", ".join(self.app.routes.keys())}, ) return section def _generate_subsections(self, section: DocumentSection): """ Generate subsections :param section: The section :type section: DocumentSection """ subsections = [] for path, data in self.pages.items(): subsections.append( RoutesSubsection( path, { "Route": f'Methods: {data["methods"]}\n\nReturn type: {data["return_type"]}', "Extra": f'Extra: {"\n".join([f" + {key}: {value}" for key, value in data["extra"].items()])}', }, section, ) ) for subsection in subsections: section.link_new_subsection(subsection) def documentate_route( self, page_path: str, return_type: Any, params: dict, methods: list, extra: dict = {}, ) -> Callable: """ Add routed page to documentation :param page_path: The page path :type page_path: str :param return_type: The return type :type return_type: Any :param params: The parameters :type params: dict :param methods: The methods :type methods: list :param extra: The extra :type extra: dict :returns: wrapper handler :rtype: Callable """ if page_path in self.pages: return def wrapper(handler): """ Wrapper for handler :param handler: The handler :type handler: callable :returns: handler :rtype: callable """ self.pages[page_path] = { "page_path": page_path, "doc": handler.__doc__, "funcname": handler.__name__, "return_type": return_type, "params": params, "methods": methods, "extra": extra, } return handler return wrapper
To add a route to the documentation, simply add the documentate_route decorator to the desired handler, something like this:
@projdoc.documentate_route('/book', str, {}, ['GET', 'POST'])
Actually, the documentation sections are routes.
❯ Generating API documentation
API generation will occur in two stages: generation of the OpenAPI specification and generation of an html template for it.
The OpenAPI specification (OAS, OpenAPI Specification) defines a formalized standard that describes the interface to the REST API service and allows you to define the capabilities of a REST service without access to its source code or documentation.
Specification 3.0.0 in more detail.
In our code it will look something like this:
spec = { "openapi": "3.0.0", "info": { "title": self._app.app_name, "version": self._app.settings.VERSION, "description": self._app.settings.DESCRIPTION, }, "paths": { }, }
In paths we will add paths that will be taken from route handlers.
Code for generating the specification
class APIDocumentation: """ This class describes an API documentation. """ def __init__(self, app: "EchoNext"): """ Constructs a new instance. :param app: The application :type app: EchoNext """ self._app = app def init_app(self, app: "EchoNext"): """ Initializes the application. :param app: The application :type app: EchoNext """ self._app = app def generate_spec(self) -> str: """ Generate OpenAPI specficiation from app routes&views :returns: jsonfied openAPI API specification :rtype: str """ spec = { "openapi": "3.0.0", "info": { "title": self._app.app_name, "version": self._app.settings.VERSION, "description": self._app.settings.DESCRIPTION, }, "paths": {}, } for url in self._app.urls: spec["paths"][url.url] = { "get": { "summary": str(f'{url.view.__doc__}. {url.view.get.__doc__}').replace('\n', '<br>') .strip(), "responses": {"200": {"description": "Successful response"}, "405": {"description": "Method not allow"}}, }, "post": { "summary": str(f'{url.view.__doc__}. {url.view.post.__doc__}').replace('\n', '<br>') .strip(), "responses": {"200": {"description": "Successful response"}, "405": {"description": "Method not allow"}}, } } for path, handler in self._app.routes.items(): spec["paths"][path] = { "get": { "summary": str(handler.__doc__) .strip() .replace("\n", ".") .replace("\t", ";"), "responses": {"200": {"description": "Successful response"}, "405": {"description": "Method not allow"}}, }, "post": { "summary": str(handler.__doc__) .strip() .replace("\n", ".") .replace("\t", ";"), "responses": {"200": {"description": "Successful response"}, "405": {"description": "Method not allow"}}, } } return spec
And in order for us to be able to view it as a web page, we will create an html template generator
HTML Template Generator
class APIDocUI: """ This class describes an api document ui. """ def __init__(self, specification: dict): """ Constructs a new instance. :param specification: The specification :type specification: dict """ self.specification = specification def generate_section(self, route: str, summary_get: str, summary_post: str, get_responses: dict, post_responses: dict) -> str: """ generate section :param route: The route :type route: str :param summary_get: The summary get :type summary_get: str :param summary_post: The summary post :type summary_post: str :param get_responses: The get responses :type get_responses: dict :param post_responses: The post responses :type post_responses: dict :returns: template section :rtype: str """ template = f''' <div class="section"> <div class="section-header"> <span>{route}</span> <span class="collapse-icon">➡️</span> </div> <div class="section-content"> <div class="method"> <strong>GET</strong> <p>{summary_get}</p> <div class="responses"> {"".join([f"<div class='response-item'>{key}: {value["description"]}.</div>" for key, value in get_responses.items()])} </div> </div> <div class="method"> <strong>POST</strong> <p>{summary_post}</p> <div class="responses"> <div class="responses"> {"".join([f"<div class='response-item'>{key}: {value["description"]}.</div>" for key, value in post_responses.items()])} </div> </div> </div> </div> </div> ''' return template def generate_html_page(self) -> str: """ Generate html page template :returns: template :rtype: str """ template = ''' <!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <title>API Documentation</title> <style> body { font-family: Arial, sans-serif; margin: 0; padding: 0; background-color: #f9f9f9; color: #333; } h1, h2, h3 { margin: 0; padding: 10px 0; } .container { max-width: 800px; margin: 40px auto; padding: 20px; background: #fff; border-radius: 8px; box-shadow: 0 2px 15px rgba(0, 0, 0, 0.1); } .version { font-size: 14px; color: #555; margin-bottom: 20px; } .info-section { border-bottom: 1px solid #ddd; padding-bottom: 20px; margin-bottom: 20px; } .section { border-radius: 5px; overflow: hidden; margin-bottom: 20px; transition: box-shadow 0.3s ease; } .section-header { padding: 15px; background: #007bff; color: white; cursor: pointer; position: relative; font-weight: bold; display: flex; justify-content: space-between; align-items: center; } .section-content { padding: 15px; display: none; overflow: hidden; background-color: #f1f1f1; } .method { border-bottom: 1px solid #ddd; padding: 10px 0; } .method:last-child { border-bottom: none; } .responses { margin-top: 10px; padding-left: 15px; font-size: 14px; color: #555; } .response-item { margin-bottom: 5px; } .collapse-icon { transition: transform 0.3s; } .collapse-icon.collapsed { transform: rotate(90deg); } .section:hover { box-shadow: 0 4px 20px rgba(0, 0, 0, 0.15); } </style> </head> <body> <div class="container"> <h1>OpenAPI Documentation</h1> <h2>PyEchoNext Web Application</h2> <div class="version">OpenAPI Version: {{openapi-version}}</div> <div class="info-section"> <h2>Application Information</h2> <p><strong>Title:</strong> {{info_title}}</p> <p><strong>Version:</strong> {{info_version}}</p> <p><strong>Description:</strong> {{info_description}}</p> </div> {{sections}} <script> document.querySelectorAll('.section-header').forEach(header => { header.addEventListener('click', () => { const content = header.nextElementSibling; const icon = header.querySelector('.collapse-icon'); if (content.style.display === "block") { content.style.display = "none"; icon.classList.add('collapsed'); } else { content.style.display = "block"; icon.classList.remove('collapsed'); } }); }); </script> </body> </html> ''' content = { '{{openapi-version}}': self.specification['openapi'], "{{info_title}}": self.specification["info"]["title"], "{{info_version}}": self.specification["info"]["version"], "{{info_description}}": self.specification["info"]["description"], "{{sections}}": "\n".join([self.generate_section(path, value['get']['summary'], value['post']['summary'], value['get']['responses'], value['post']['responses']) for path, value in self.specification["paths"].items()]) } for key, value in content.items(): template = template.replace(key, value) return template
❯ Configuration and loading of settings
Our application will need configuration such as meta information or setting up directories for operation and other things.
I also decided to make the configuration universal - it can be loaded from .ini, environment variables or a python file.
The settings class itself will be passed to the constructor of our future application class. It looks like this:
@dataclass class Settings: """ This class describes settings. """ BASE_DIR: str TEMPLATES_DIR: str SECRET_KEY: str VERSION: str = "1.0.0" DESCRIPTION: str = "Echonext webapp" LOCALE: str = "DEFAULT" LOCALE_DIR: str = None
BASE_DIR - base project directory
TEMPLATES_DIR - html template directory
SECRET_KEY - secret key
VERSION - version
DESCRIPTION - description
LOCALE - localization code
LOCALE_DIR - directory with localization files.
We will use configparser to load .ini, python-dotenv for environment variables, and importlib for python files.
Source code config.py
import os import importlib from pathlib import Path from dataclasses import dataclass from enum import Enum from configparser import ConfigParser from dotenv import load_dotenv def dynamic_import(module: str): """ Dynamic import with importlib :param module: The module :type module: str :returns: module :rtype: module """ return importlib.import_module(str(module)) @dataclass class Settings: """ This class describes settings. """ BASE_DIR: str TEMPLATES_DIR: str SECRET_KEY: str VERSION: str = "1.0.0" DESCRIPTION: str = "Echonext webapp" LOCALE: str = "DEFAULT" LOCALE_DIR: str = None class SettingsConfigType(Enum): """ This class describes a settings configuration type. """ INI = "ini" DOTENV = "dotenv" PYMODULE = "pymodule" class SettingsLoader: """ This class describes a settings loader. """ def __init__(self, config_type: SettingsConfigType, filename: str = None): """ Constructs a new instance. :param config_type: The configuration type :type config_type: SettingsConfigType :param filename: The filename :type filename: str """ self.config_type: SettingsConfigType = config_type self.filename: str = filename self.filename: Path = Path(self.filename) if not self.filename.exists(): raise FileNotFoundError(f'Config file "{self.filename}" don\'t exists.') def _load_ini_config(self) -> dict: """ Loads a .ini config file :returns: config dictionary :rtype: dict """ config = ConfigParser() config.read(self.filename) return config["Settings"] def _load_env_config(self) -> dict: """ Loads an environment configuration. :returns: config dictionary :rtype: dict """ load_dotenv(self.filename) config = { "BASE_DIR": os.environ.get("PEN_BASE_DIR"), "TEMPLATES_DIR": os.environ.get("PEN_TEMPLATES_DIR"), "SECRET_KEY": os.environ.get("PEN_SECRET_KEY"), "LOCALE": os.environ.get("PEN_LOCALE", "DEFAULT"), "LOCALE_DIR": os.environ.get("PEN_LOCALE_DIR", None), "VERSION": os.environ.get("PEN_VERSION", "1.0.0"), "DESCRIPTION": os.environ.get("PEN_DESCRIPTION", "EchoNext webapp"), } return config def _load_pymodule_config(self) -> dict: """ Loads a pymodule configuration. :returns: config dictionary :rtype: dict """ config_module = dynamic_import(str(self.filename).replace(".py", "")) return { "BASE_DIR": config_module.BASE_DIR, "TEMPLATES_DIR": config_module.TEMPLATES_DIR, "SECRET_KEY": config_module.SECRET_KEY, "LOCALE": config_module.LOCALE, "LOCALE_DIR": config_module.LOCALE_DIR, "VERSION": config_module.VERSION, "DESCRIPTION": config_module.DESCRIPTION, } def get_settings(self) -> Settings: """ Gets the settings. :returns: The settings. :rtype: Settings """ if self.config_type == SettingsConfigType.INI: self.config = self._load_ini_config() elif self.config_type == SettingsConfigType.DOTENV: self.config = self._load_env_config() elif self.config_type == SettingsConfigType.PYMODULE: self.config = self._load_pymodule_config() return Settings( BASE_DIR=self.config.get("BASE_DIR", "."), TEMPLATES_DIR=self.config.get("TEMPLATES_DIR", "templates"), SECRET_KEY=self.config.get("SECRET_KEY", ""), LOCALE=self.config.get("LOCALE", "DEFAULT"), LOCALE_DIR=self.config.get("LOCALE_DIR", None), VERSION=self.config.get("VERSION", "1.0.0"), DESCRIPTION=self.config.get("DESCRIPTION", "EchoNext webapp"), )
Examples of config loading:
DOTENV
config_loader = SettingsLoader(SettingsConfigType.DOTENV, 'example_env') settings = config_loader.get_settings()
example_env file:
PEN_BASE_DIR=. PEN_TEMPLATES_DIR=templates PEN_SECRET_KEY=secret-key PEN_LOCALE=RU_RU PEN_LOCALE_DIR=locales PEN_VERSION=1.0.0 PEN_DESCRIPTION=Example
INI
config_loader = SettingsLoader(SettingsConfigType.INI, 'example_ini.ini') settings = config_loader.get_settings()
Example_ini.ini file:
[Settings] BASE_DIR=. TEMPLATES_DIR=templates SECRET_KEY=secret-key LOCALE=DEFAULT VERSION=1.0.0 DESCRIPTION=Example
PyModule
config_loader = SettingsLoader(SettingsConfigType.PYMODULE, 'example_module.py') settings = config_loader.get_settings()
Example_module.py file:
import os BASE_DIR = os.path.dirname(os.path.abspath(__file__)) TEMPLATES_DIR = 'templates' SECRET_KEY = 'secret-key' VERSION = '1.0.0' DESCRIPTION = 'Echonext webapp' LOCALE = 'DEFAULT' LOCALE_DIR = None
Render html templates
I decided to have a small built-in engine and integrate Jinja2.
Let's start with the built-in one. It will be based on regex expressions. At the moment I have implemented two:
FOR_BLOCK_PATTERN = re.compile( r"{% for (?P<variable>[a-zA-Z]+) in (?P<seq>[a-zA-Z]+) %}(?P<content>[\S\s]+)(?={% endfor %}){% endfor %}" ) VARIABLE_PATTERN = re.compile(r"{{ (?P<variable>[a-zA-Z_]+) }}")
It is similar to Jinja2. For a for loop you need to use the construction {% for ... in ... %}{% endfor %}, and to display variables {{ <переменная> }}.
The directory with templates will be taken from the settings class.
A function will be created to generate render_template(request: Request, template_name: str, **kwargs). It needs Request, the name of the template (without the directory), and also the context - that is, kwargs. That is, when calling render_template(request, 'index.html', name="Vasya") in the template you can use the name variable.
Built-in template engine
import os import re from loguru import logger from pyechonext.request import Request from pyechonext.utils.exceptions import TemplateNotFileError FOR_BLOCK_PATTERN = re.compile( r"{% for (?P<variable>[a-zA-Z]+) in (?P<seq>[a-zA-Z]+) %}(?P<content>[\S\s]+)(?={% endfor %}){% endfor %}" ) VARIABLE_PATTERN = re.compile(r"{{ (?P<variable>[a-zA-Z_]+) }}") class TemplateEngine: """ This class describes a built-in template engine. """ def __init__(self, base_dir: str, templates_dir: str): """ Constructs a new instance. :param base_dir: The base dir :type base_dir: str :param templates_dir: The templates dir :type templates_dir: str """ self.templates_dir = os.path.join(base_dir, templates_dir) def _get_template_as_string(self, template_name: str) -> str: """ Gets the template as string. :param template_name: The template name :type template_name: str :returns: The template as string. :rtype: str :raises TemplateNotFileError: Template is not a file """ template_name = os.path.join(self.templates_dir, template_name) if not os.path.isfile(template_name): raise TemplateNotFileError(f'Template "{template_name}" is not a file') with open(template_name, "r") as file: content = file.read() return content def _build_block_of_template(self, context: dict, raw_template_block: str) -> str: """ Builds a block of template. :param context: The context :type context: dict :param raw_template_block: The raw template block :type raw_template_block: str :returns: The block of template. :rtype: str """ used_vars = VARIABLE_PATTERN.findall(raw_template_block) if used_vars is None: return raw_template_block for var in used_vars: var_in_template = "{{ %s }}" % (var) processed_template_block = re.sub( var_in_template, str(context.get(var, "")), raw_template_block ) return processed_template_block def _build_statement_for_block(self, context: dict, raw_template_block: str) -> str: """ Build statement `for` block :param context: The context :type context: dict :param raw_template_block: The raw template block :type raw_template_block: str :returns: The statement for block. :rtype: str """ statement_for_block = FOR_BLOCK_PATTERN.search(raw_template_block) if statement_for_block is None: return raw_template_block builded_statement_block_for = "" for variable in context.get(statement_for_block.group("seq"), []): builded_statement_block_for += self._build_block_of_template( {**context, statement_for_block.group("variable"): variable}, statement_for_block.group("content"), ) processed_template_block = FOR_BLOCK_PATTERN.sub( builded_statement_block_for, raw_template_block ) return processed_template_block def build(self, context: dict, template_name: str) -> str: """ Build template :param context: The context :type context: dict :param template_name: The template name :type template_name: str :returns: raw template string :rtype: str """ raw_template = self._get_template_as_string(template_name) processed_template = self._build_statement_for_block(context, raw_template) return self._build_block_of_template(context, processed_template) def render_template(request: Request, template_name: str, **kwargs) -> str: """ Render template :param request: The request :type request: Request :param template_name: The template name :type template_name: str :param kwargs: The keywords arguments :type kwargs: dictionary :returns: raw template string :rtype: str :raises AssertionError: BASE_DIR and TEMPLATES_DIR is empty """ logger.warn( "Built-in template engine is under development and may be unstable or contain bugs" ) assert request.settings.BASE_DIR assert request.settings.TEMPLATES_DIR engine = TemplateEngine(request.settings.BASE_DIR, request.settings.TEMPLATES_DIR) context = kwargs logger.debug(f"Built-in template engine: render {template_name} ({request.path})") return engine.build(context, template_name)
For Jinja2 there will be very similar code so that there are no problems with support.
Jinja2 integration code
from os.path import join, exists, getmtime from jinja2 import BaseLoader, TemplateNotFound from jinja2 import Environment, select_autoescape from loguru import logger from pyechonext.request import Request class TemplateLoader(BaseLoader): """ This class describes a jinja2 template loader. """ def __init__(self, path: str): """ Constructs a new instance. :param path: The path :type path: str """ self.path = path def get_source(self, environment, template): path = join(self.path, template) if not exists(path): raise TemplateNotFound(template) mtime = getmtime(path) with open(path) as f: source = f.read() return source, path, lambda: mtime == getmtime(path) class TemplateEngine: """ This class describes a jinja template engine. """ def __init__(self, base_dir: str, templates_dir: str): """ Constructs a new instance. :param base_dir: The base dir :type base_dir: str :param templates_dir: The templates dir :type templates_dir: str """ self.base_dir = base_dir self.templates_dir = join(base_dir, templates_dir) self.env = Environment( loader=TemplateLoader(self.templates_dir), autoescape=select_autoescape() ) def build(self, template_name: str, **kwargs): template = self.env.get_template(template_name) return template.render(**kwargs) def render_template(request: Request, template_name: str, **kwargs) -> str: """ Render template :param request: The request :type request: Request :param template_name: The template name :type template_name: str :param kwargs: The keywords arguments :type kwargs: dictionary :returns: raw template string :rtype: str :raises AssertionError: BASE_DIR and TEMPLATES_DIR is empty """ assert request.settings.BASE_DIR assert request.settings.TEMPLATES_DIR engine = TemplateEngine(request.settings.BASE_DIR, request.settings.TEMPLATES_DIR) logger.debug(f"Jinja2 template engine: render {template_name} ({request.path})") return engine.build(template_name, **kwargs)
❯ Response-request
Whether you are in Rybatskino or southern Broms. If there is a request, then there is a response.
In computer science, request-response or request-replica is one of the basic methods used by computers to communicate with each other on a network, in which the first computer sends a request for some data and the second one responds to the request. More specifically, it is a messaging pattern in which a requester sends a request message to a responder system, which receives and processes the request, ultimately returning a message in response. This is similar to a telephone call, in which the caller must wait for the recipient to pick up the phone before anything can be discussed.
❯ Request
Request is a request that contains data for interaction between the client and the API: base URL, endpoint, method used, headers, etc.
The class itself looks like this:
class Request: """ This class describes a request. """ def __init__(self, environ: dict, settings: Settings): """ Constructs a new instance. :param environ: The environ :type environ: dict """ self.environ: dict = environ self.settings: Settings = settings self.method: str = self.environ["REQUEST_METHOD"] self.path: str = self.environ["PATH_INFO"] self.GET: dict = self._build_get_params_dict(self.environ["QUERY_STRING"]) self.POST: dict = self._build_post_params_dict(self.environ["wsgi.input"].read()) self.user_agent: str = self.environ["HTTP_USER_AGENT"] self.extra: dict = {} logger.debug(f"New request created: {self.method} {self.path}") def __getattr__(self, item: Any) -> Union[Any, None]: """ Magic method for get attrs (from extra) :param item: The item :type item: Any :returns: Item from self.extra or None :rtype: Union[Any, None] """ return self.extra.get(item, None) def _build_get_params_dict(self, raw_params: str): """ Builds a get parameters dictionary. :param raw_params: The raw parameters :type raw_params: str """ return parse_qs(raw_params) def _build_post_params_dict(self, raw_params: bytes): """ Builds a post parameters dictionary. :param raw_params: The raw parameters :type raw_params: bytes """ return parse_qs(raw_params.decode())
Request requires the following arguments to create:
environ (dictionary) - web environment (generated by gunicorn).
settings (dataclass object pyechonext.config.Settings).
Request has the following public attributes:
environ (dictionary) - web environment.
settings (dataclass object pyechonext.config.Settings).
method (string) - http method.
path (string) - path.
GET (dictionary) - get request parameters.
POST (dictionary) - post request parameters.
user_agent (string) — User-Agent.
extra (dictionary) - additional parameters (for example, for middleware).
Request also has the following methods:
__getattr__- magic descriptor method for getting attributes (to get elements from the extra attribute)._build_get_params_dict— private method for parsing get request parameters._build_post_params_dict— private method for parsing post request parameters.
❯ Response
Response is the response that contains the data returned by the server, including content, status code, and headers.
import json from typing import Dict, Iterable, Union, Any, List, Tuple, Optional from socks import method from loguru import logger from pyechonext.request import Request class Response: """ This dataclass describes a response. """ default_content_type: str = "text/html" default_charset: str = "UTF-8" unicode_errors: str = "strict" default_conditional_response: bool = False default_body_encoding: str = "UTF-8" def __init__( self, request: Request, use_i18n: bool = False, status_code: Optional[int] = 200, body: Optional[str] = None, headers: Optional[Dict[str, str]] = {}, content_type: Optional[str] = None, charset: Optional[str] = None, **kwargs, ): """ Constructs a new instance. :param request: The request :type request: Request :param use_i18n: The use i 18 n :type use_i18n: bool :param status_code: The status code :type status_code: int :param body: The body :type body: str :param headers: The headers :type headers: Dict[str, str] :param content_type: The content type :type content_type: str :param charset: The charset :type charset: str :param kwargs: The keywords arguments :type kwargs: dictionary """ if status_code == 200: self.status_code: str = "200 OK" else: self.status_code: str = str(status_code) if content_type is None: self.content_type: str = self.default_content_type else: self.content_type: str = content_type if charset is None: self.charset: str = self.default_charset else: self.charset: str = charset if body is not None: self.body: str = body else: self.body: str = "" self._headerslist: list = headers self._added_headers: list = [] self.request: Request = request self.extra: dict = {} self.use_i18n: bool = use_i18n self.i18n_kwargs = kwargs self._update_headers() def __getattr__(self, item: Any) -> Union[Any, None]: """ Magic method for get attrs (from extra) :param item: The item :type item: Any :returns: Item from self.extra or None :rtype: Union[Any, None] """ return self.extra.get(item, None) def _structuring_headers(self, environ): headers = { "Host": environ["HTTP_HOST"], "Accept": environ["HTTP_ACCEPT"], "User-Agent": environ["HTTP_USER_AGENT"], } for name, value in headers.items(): self._headerslist.append((name, value)) for header_tuple in self._added_headers: self._headerslist.append(header_tuple) def _update_headers(self) -> None: """ Sets the headers by environ. :param environ: The environ :type environ: dict """ self._headerslist = [ ("Content-Type", f"{self.content_type}; charset={self.charset}"), ("Content-Length", str(len(self.body))), ] def add_headers(self, headers: List[Tuple[str, str]]): """ Adds new headers. :param headers: The headers :type headers: List[Tuple[str, str]] """ for header in headers: self._added_headers.append(header) def _encode_body(self): """ Encodes a body. """ if self.content_type.split("/")[-1] == "json": self.body = str(self.json) try: self.body = self.body.encode("UTF-8") except AttributeError: self.body = str(self.body).encode("UTF-8") def __call__(self, environ: dict, start_response: method) -> Iterable: """ Makes the Response object callable. :param environ: The environ :type environ: dict :param start_response: The start response :type start_response: method :returns: response body :rtype: Iterable """ self._encode_body() self._update_headers() self._structuring_headers(environ) logger.debug( f"[{environ['REQUEST_METHOD']} {self.status_code}] Run response: {self.content_type}" ) start_response(status=self.status_code, headers=self._headerslist) return iter([self.body]) @property def json(self) -> dict: """ Parse request body as JSON. :returns: json body :rtype: dict """ if self.body: if self.content_type.split("/")[-1] == "json": return json.dumps(self.body) else: return json.dumps(self.body.decode("UTF-8")) return {} def __repr__(self): """ Returns a unambiguous string representation of the object (for debug...). :returns: String representation of the object. :rtype: str """ return f"<{self.__class__.__name__} at 0x{abs(id(self)):x} {self.status_code}>"
Response has the following arguments:
request (request class object) - request.
[optional] status_code (integer value) — status code of the response.
[optional] body (string) — the body of the response.
[optional] headers (dictionary) — response headers.
[optional] content_type (string) — content type of the response.
[optional] charset (string) — response encoding.
[optional] use_i18n (boolean value) — whether to use i18n (default False).
Response has the following attributes:
status_code (string) — status code (default "200 OK").
content_type (string) — content type (defaults to default_content_type).
charset (string) - encoding (defaults to default_charset).
body (string) — body of the answer (defaults to the empty string).
_headerslist(list) - private list of response headers._added_headers(list) - private list of added response headers.request (request class object) - request.
extra (dictionary) - additional parameters.
Response has the following methods:
__getattr__- magic descriptor method for getting attributes (to get elements from the extra attribute)._structuring_headers- a private method for structuring headers from the web environment._update_headers— private method for updating (rewriting) header lists.add_headers- public method for adding headers._encode_body— encoding of the response body.__call__— a magic method that makes the Response object callable.json— class property for receiving the response body in the form of json.
❯ Views (handlers)
View is an abstraction of the site route (django-like). It must have two methods: get And post (to respond to get and post requests). These methods should return:
Data, page content. This can be a dictionary or a string.
OR:
Response class object (pyechonext.response)
View is an object of the View class:
class View(ABC): """ Page view """ @abstractmethod def get( self, request: Request, response: Response, *args, **kwargs ) -> Union[Response, Any]: """ Get :param request: The request :type request: Request :param response: The response :type response: Response :param args: The arguments :type args: list :param kwargs: The keywords arguments :type kwargs: dictionary """ raise NotImplementedError @abstractmethod def post( self, request: Request, response: Response, *args, **kwargs ) -> Union[Response, Any]: """ Post :param request: The request :type request: Request :param response: The response :type response: Response :param args: The arguments :type args: list :param kwargs: The keywords arguments :type kwargs: dictionary """ raise NotImplementedError
And let me show an example View:
class IndexView(View): def get( self, request: Request, response: Response, **kwargs ) -> Union[Response, Any]: """ Get :param request: The request :type request: Request :param response: The response :type response: Response :param args: The arguments :type args: list :param kwargs: The keywords arguments :type kwargs: dictionary """ return "Welcome to pyEchoNext webapplication!" def post( self, request: Request, response: Response, **kwargs ) -> Union[Response, Any]: """ Post :param request: The request :type request: Request :param response: The response :type response: Response :param args: The arguments :type args: list :param kwargs: The keywords arguments :type kwargs: dictionary """ return "Message has accepted!"
❯ URLS
In order to connect Views to the application, we will create an abstraction layer - a URL dataclass, which will contain the path and the View class itself. Moreover, View must be passed without creating an object, that is, the class itself.
from dataclasses import dataclass from typing import Type from pyechonext.views import View, IndexView @dataclass class URL: """ This dataclass describes an url. """ url: str view: Type[View] url_patterns = [URL(url="/", view=IndexView)]
url_patterns - built-in patterns. For example, we use the previously created IndexView.
❯ Middleware (middleware)
So, to implement, for example, a cookie, we will need to work with request response while the server is running. To do this, we will use a middleware abstraction.
class BaseMiddleware(ABC): """ This abstract class describes a base middleware. """ @abstractmethod def to_request(self, request: Request): """ To request method :param request: The request :type request: Request """ raise NotImplementedError @abstractmethod def to_response(self, response: Response): """ To response method :param response: The response :type response: Response """ raise NotImplementedError
It has two abstract methods - to_request and to_response.
Let's implement a basic session Middleware to add cookies:
class SessionMiddleware(BaseMiddleware): """ This class describes a session (cookie) middleware. """ def to_request(self, request: Request): """ Set to request :param request: The request :type request: Request """ cookie = request.environ.get("HTTP_COOKIE", None) if not cookie: return session_id = parse_qs(cookie)["session_id"][0] logger.debug( f"Set session_id={session_id} for request {request.method} {request.path}" ) request.extra["session_id"] = session_id def to_response(self, response: Response): """ Set to response :param response: The response :type response: Response """ if not response.request.session_id: session_id = uuid4() logger.debug( f"Set session_id={session_id} for response {response.status_code} {response.request.path}" ) response.add_headers( [ ("Set-Cookie", f"session_id={session_id}"), ] ) middlewares = [SessionMiddleware] # Список мидлварей
And now let's move on to app.py itself - the application.
❯ Utilities
We need to create a file utils/__init__.py, which will contain a small helper function _prepare_url. It will trim the URL from everything unnecessary:
from datetime import datetime def get_current_datetime() -> str: """ Gets the current datetime. :returns: The current datetime. :rtype: str """ date = datetime.now() return date.strftime("%Y-%m-%d %H:%M:%S") def _prepare_url(url: str) -> str: """ Prepare URL (remove ending /) :param url: The url :type url: str :returns: prepared url :rtype: str """ try: if url[-1] == "/" and len(url) > 1: return url[:-1] except IndexError: return "/" return url
❯ Application
The basis is the EchoNext class (pyechonext.app).
Let's create it.
We import all the necessary modules:
import inspect from enum import Enum from typing import Iterable, Callable, List, Type, Tuple, Optional, Union from dataclasses import dataclass from socks import method from parse import parse from loguru import logger from pyechonext.urls import URL from pyechonext.views import View from pyechonext.request import Request from pyechonext.response import Response from pyechonext.utils.exceptions import ( RoutePathExistsError, MethodNotAllow, URLNotFound, WebError, TeapotError, ) from pyechonext.utils import _prepare_url from pyechonext.config import Settings from pyechonext.middleware import BaseMiddleware from pyechonext.i18n_l10n.i18n import JSONi18nLoader from pyechonext.i18n_l10n.l10n import JSONLocalizationLoader
Let's create an application type dataclass:
class ApplicationType(Enum): """ This enum class describes an application type. """ JSON = "application/json" HTML = "text/html" PLAINTEXT = "text/plain" TEAPOT = "server/teapot"
JSON - mainly for APIs.
HTML - for a full-fledged website.
PLAINTEXT - just text.
Next, we will create a HistoryEntry dataclass to store the history of requests and responses:
@dataclass class HistoryEntry: request: Request response: Response
Let's start creating the application class:
class EchoNext: """ This class describes an EchoNext WSGI Application. """ __slots__ = ( "app_name", "settings", "middlewares", "application_type", "urls", "routes", "i18n_loader", "l10n_loader", "history", )
__slots__ are slots (class attributes are listed in a tuple). This is a mechanism that allows you to optimize memory usage and speed up access to class attributes. When you create an object of a class in Python, the interpreter allocates memory to store all the attributes of that object.
After that, let's create a magic class constructor method:
def __init__( self, app_name: str, settings: Settings, middlewares: List[Type[BaseMiddleware]], urls: Optional[List[URL]] = [], application_type: Optional[ApplicationType] = ApplicationType.JSON, ): """ Constructs a new instance. :param app_name: The application name :type app_name: str :param settings: The settings :type settings: Settings :param middlewares: The middlewares :type middlewares: List[BaseMiddleware] :param urls: The urls :type urls: List[URL] :param application_type: The application type :type application_type: Optional[ApplicationType] """ self.app_name = app_name self.settings = settings self.middlewares = middlewares self.application_type = application_type self.routes = {} self.urls = urls self.history: List[HistoryEntry] = [] self.i18n_loader = JSONi18nLoader( self.settings.LOCALE, self.settings.LOCALE_DIR ) self.l10n_loader = JSONLocalizationLoader( self.settings.LOCALE, self.settings.LOCALE_DIR ) logger.debug(f"Application {self.application_type.value}: {self.app_name}") if self.application_type == ApplicationType.TEAPOT: raise TeapotError("Where's my coffie?")
Let's look at the attributes:
app_name — application name.
settings — an instance of the Settings dataclass.
middlewares — list of middlewares.
application_type — application type.
routes — a dictionary with routes that were specified through the route_page decorator (flask-like path, we’ll look at it later).
urls — list of URLs (for View integration).
history - a list from HistoryEntry. Request-response history.
i18n_loader - i18n loader.
l10n_loader - l10n loader.
Let's implement the following method:
def _find_view(self, raw_url: str) -> Union[Type[URL], None]: """ Finds a view by raw url. :param raw_url: The raw url :type raw_url: str :returns: URL dataclass :rtype: Type[URL] """ url = _prepare_url(raw_url) for path in self.urls: if url == _prepare_url(path.url): return path return None
It is needed to find a view using a raw URL. If it is found, return the URL, otherwise None.
Let's create a method _check_request_method:
def _check_request_method(self, view: View, request: Request): """ Check request method for view :param view: The view :type view: View :param request: The request :type request: Request :raises MethodNotAllow: Method not allow """ if not hasattr(view, request.method.lower()): raise MethodNotAllow(f"Method not allow: {request.method}")
This method simply checks if the method is available in the View.
def _get_view(self, request: Request) -> View: """ Gets the view. :param request: The request :type request: Request :returns: The view. :rtype: View """ url = request.path return self._find_view(url)
The method above gets the View request path.
The following two methods generate the request and response:
def _get_request(self, environ: dict) -> Request: """ Gets the request. :param environ: The environ :type environ: dict :returns: The request. :rtype: Request """ return Request(environ, self.settings) def _get_response(self, request: Request) -> Response: """ Gets the response. :returns: The response. :rtype: Response """ return Response(request, content_type=self.application_type.value)
Now let’s implement the same route_page decorator:
def route_page(self, page_path: str) -> Callable: """ Creating a New Page Route :param page_path: The page path :type page_path: str :returns: wrapper handler :rtype: Callable """ if page_path in self.routes: raise RoutePathExistsError("Such route already exists.") def wrapper(handler): """ Wrapper for handler :param handler: The handler :type handler: callable :returns: handler :rtype: callable """ self.routes[page_path] = handler return handler return wrapper
Now let's create two methods for applying middleware to a request:
def _apply_middleware_to_request(self, request: Request): """ Apply middleware to request :param request: The request :type request: Request """ for middleware in self.middlewares: middleware().to_request(request) def _apply_middleware_to_response(self, response: Response): """ Apply middleware to response :param response: The response :type response: Response """ for middleware in self.middlewares: middleware().to_response(response)
Let's implement the default response method. That is, we will assign, for example, a 404 code to the response if the page is not found:
def _default_response(self, response: Response, error: WebError) -> None: """ Get default response (404) :param response: The response :type response: Response """ response.status_code = str(error.code) response.body = str(error)
Now let's implement a method for finding a handler. By the way, my Views have higher priority than routes:
def _find_handler(self, request: Request) -> Tuple[Callable, str]: """ Finds a handler. :param request_path: The request path :type request_path: str :returns: handler function and parsed result :rtype: Tuple[Callable, str] """ url = _prepare_url(request.path) for path, handler in self.routes.items(): parse_result = parse(path, url) if parse_result is not None: return handler, parse_result.named view = self._get_view(request) if view is not None: parse_result = parse(view.url, url) if parse_result is not None: return view.view, parse_result.named return None, None
Let's create an on-the-fly localization switch method:
def switch_locale(self, locale: str, locale_dir: str): """ Switch to another locale i18n :param locale: The locale :type locale: str :param locale_dir: The locale dir :type locale_dir: str """ logger.info(f"Switch to another locale: {locale_dir}/{locale}") self.i18n_loader.locale = locale self.i18n_loader.directory = locale_dir self.i18n_loader.translations = self.i18n_loader.load_locale( self.i18n_loader.locale, self.i18n_loader.directory ) self.l10n_loader.locale = locale self.l10n_loader.directory = directory self.i18n_loader.locale_settings = self.l10n_loader.load_locale( self.l10n_loader.locale, self.l10n_loader.directory )
Now let’s create a request handler that will process everything, including finding and generating errors.
def _handle_request(self, request: Request) -> Response: """ Handle response from request :param request: The request :type request: Request :returns: Response callable object :rtype: Response """ logger.debug(f"Handle request: {request.path}") response = self._get_response(request) handler, kwargs = self._find_handler(request) if handler is not None: if inspect.isclass(handler): handler = getattr(handler(), request.method.lower(), None) if handler is None: raise MethodNotAllow(f"Method not allowed: {request.method}") result = handler(request, response, **kwargs) if isinstance(result, Response): response = result if response.use_i18n: response.body = self.i18n_loader.get_string( response.body, **response.i18n_kwargs ) else: response.body = self.i18n_loader.get_string(result) if not response.use_i18n: response.body = result else: raise URLNotFound(f'URL "{request.path}" not found.') return response
And finally, the method __call__. It will make our class callable.
def __call__(self, environ: dict, start_response: method) -> Iterable: """ Makes the application object callable :param environ: The environ :type environ: dict :param start_response: The start response :type start_response: method :returns: response body :rtype: Iterable """ request = self._get_request(environ) self._apply_middleware_to_request(request) response = self._get_response(request) try: response = self._handle_request(request) self._apply_middleware_to_response(response) except URLNotFound as err: logger.error( "URLNotFound error has been raised: set default response (404)" ) self._apply_middleware_to_response(response) self._default_response(response, error=err) except MethodNotAllow as err: logger.error( "MethodNotAllow error has been raised: set default response (405)" ) self._apply_middleware_to_response(response) self._default_response(response, error=err) self.history.append(HistoryEntry(request=request, response=response)) return response(environ, start_response)
And yes, errors are processed and will notify the site user in some cases. For example URLNotFound will generate a 404 error and so on. This will also enable the developer to cause web errors in the web application code.
And in the same method the final work takes place.
❯ Examples
Let me write some examples.
❯ Simple webapp
Generation of documentation, and demonstration of registering routes in different ways.
import os from pyechonext.utils.exceptions import MethodNotAllow from pyechonext.app import ApplicationType, EchoNext from pyechonext.views import View from pyechonext.urls import URL, IndexView from pyechonext.config import SettingsLoader, SettingsConfigType from pyechonext.template_engine.jinja import render_template from pyechonext.middleware import middlewares from pyechonext.docsgen import ProjDocumentation class UsersView(View): def get(self, request, response, **kwargs): return render_template( request, "index.html", user_name="User", session_id=request.session_id, friends=["Bob", "Anna", "John"], ) def post(self, request, response, **kwargs): raise MethodNotAllow(f"Request {request.path}: method not allow") url_patterns = [URL(url="/", view=IndexView), URL(url="/users", view=UsersView)] config_loader = SettingsLoader(SettingsConfigType.PYMODULE, 'example_module.py') settings = config_loader.get_settings() echonext = EchoNext( __name__, settings, middlewares, urls=url_patterns, application_type=ApplicationType.HTML, ) apidoc = ProjDocumentation(echonext) @echonext.route_page("/book") @apidoc.documentate_route('/book', str, {}, ['GET', 'POST']) class BooksResource(View): """ This class describes a books resource. """ def get(self, request, response, **kwargs): """ get queries :param request: The request :type request: Request :param response: The response :type response: Response :param kwargs: The keywords arguments :type kwargs: dictionary :returns: result :rtype: str """ return f"GET Params: {request.GET}" def post(self, request, response, **kwargs): """ post queries :param request: The request :type request: Request :param response: The response :type response: Response :param kwargs: The keywords arguments :type kwargs: dictionary :returns: result :rtype: str """ return f"POST Params: {request.POST}" apidoc.generate_documentation()
To do this you need a templates/index.html file and an example_module.py file.
example_module.py is the settings file:
import os BASE_DIR = os.path.dirname(os.path.abspath(__file__)) TEMPLATES_DIR = 'templates' SECRET_KEY = 'secret-key' LOCALE = 'DEFAULT' LOCALE_DIR = None VERSION = 0.1.0 DESCRIPTION = 'Example echonext webapp'
❯ Localization and docs-api ui
import os from pyechonext.utils.exceptions import MethodNotAllow from pyechonext.app import ApplicationType, EchoNext from pyechonext.views import View from pyechonext.urls import URL, IndexView from pyechonext.config import SettingsLoader, SettingsConfigType from pyechonext.response import Response from pyechonext.template_engine.jinja import render_template from pyechonext.middleware import middlewares from pyechonext.docsgen import ProjDocumentation from pyechonext.apidoc_ui import APIDocumentation, APIDocUI class UsersView(View): def get(self, request, response, **kwargs): return render_template( request, "index.html", user_name="User", session_id=request.session_id, friends=["Bob", "Anna", "John"], ) def post(self, request, response, **kwargs): raise MethodNotAllow(f"Request {request.path}: method not allow") url_patterns = [URL(url="/", view=IndexView), URL(url="/users", view=UsersView)] config_loader = SettingsLoader(SettingsConfigType.PYMODULE, 'el_config.py') settings = config_loader.get_settings() echonext = EchoNext( __name__, settings, middlewares, urls=url_patterns, application_type=ApplicationType.HTML, ) apidoc = APIDocumentation(echonext) projdoc = ProjDocumentation(echonext) @echonext.route_page('/api-docs') def api_docs(request, response): ui = APIDocUI(apidoc.generate_spec()) return ui.generate_html_page() @echonext.route_page("/book") @projdoc.documentate_route('/book', str, {}, ['GET', 'POST']) class BooksResource(View): """ This class describes a books resource. """ def get(self, request, response, **kwargs): """ get queries :param request: The request :type request: Request :param response: The response :type response: Response :param kwargs: The keywords arguments :type kwargs: dictionary :returns: result :rtype: str """ return echonext.l10n_loader.format_currency(1305.50) def post(self, request, response, **kwargs): """ post queries :param request: The request :type request: Request :param response: The response :type response: Response :param kwargs: The keywords arguments :type kwargs: dictionary :returns: result :rtype: str """ return echonext.i18n_loader.get_string('title %{name}', name='Localization Site') projdoc.generate_documentation()
❯ Sample application with a database
I will use my own ORM − link to repository. It is installed simply: pip3 install sqlsymphony_orm.
import os from pyechonext.app import ApplicationType, EchoNext from pyechonext.config import Settings from sqlsymphony_orm.datatypes.fields import IntegerField, RealField, TextField from sqlsymphony_orm.models.session_models import SessionModel from sqlsymphony_orm.models.session_models import SQLiteSession from pyechonext.middleware import middlewares settings = Settings( BASE_DIR=os.path.dirname(os.path.abspath(__file__)), TEMPLATES_DIR="templates", SECRET_KEY="secret" ) echonext = EchoNext( __name__, settings, middlewares, application_type=ApplicationType.HTML ) session = SQLiteSession("echonext.db") class User(SessionModel): __tablename__ = "Users" id = IntegerField(primary_key=True) name = TextField(null=False) cash = RealField(null=False, default=0.0) def __repr__(self): return f"<User {self.pk}>" @echonext.route_page("/") def home(request, response): user = User(name="John", cash=100.0) session.add(user) session.commit() return "Hello from the HOME page" @echonext.route_page("/users") def about(request, response): users = session.get_all_by_model(User) return f"Users: {[f'{user.name}: {user.cash}$' for user in users]}"
Thus, we have an almost complete framework in Python. So far he is missing:
Authentication;
Websockets;
celery integration;
Caching;
Static files.
If you liked the article, I can write a second part, where we will implement even more functionality.
❯ Conclusion
This is one of my largest and most developed projects. It was difficult, but interesting. I have a better understanding of the structure of web applications and frameworks.
If you have questions or suggestions, write in the comments, I’ll be glad to listen.
The source code repository is available at link.
I will be glad if you join my little telegram blog. Announcements of articles, news from the IT world and useful materials for studying programming and related fields. Don't hit me :)
News, product reviews and competitions from the Timeweb.Cloud team - in our Telegram channel ↩

Go ↩
Why This Matters In Practice
Beyond the original publication, How to create your own web framework in Python matters because teams need reusable decision patterns, not one-off anecdotes. Good day, habr! In this article, we will create our own web framework in Python using gunicorn. It will be lightweight and have basic functi...
Operational Takeaways
- Separate core principles from context-specific details before implementation.
- Define measurable success criteria before adopting the approach.
- Validate assumptions on a small scope, then scale based on evidence.
Quick Applicability Checklist
- Can this be reproduced with your current team and constraints?
- Do you have observable signals to confirm improvement?
- What trade-off (speed, cost, complexity, risk) are you accepting?