序幕
哈Ha!本文专门分析了大约一个星期前发布的下一个Python框架的优缺点。
所以,一个小的抒情题外话。在众所周知的事件中,当我们有点自我隔离时,我们有更多的空闲时间。有人拿出了待阅读的文学作品清单,有人开始学习另一门外语,有人继续按着多坦语写作,却没有注意变化。但是我(很抱歉,本文将包含很多“ I”,我有些as愧)决定并尝试做一些有用的事情。但是,有用性值得商bat。读者最有可能首先想到的明显问题是:“嗯,Python框架?另一个?对不起,为什么呢?毕竟我们不是JavaScript!”
实际上,这正是本文要讨论的内容:是否有必要?如有必要,给谁?与已经存在的区别是什么?它如何具有吸引力,以及为什么可以不等待第一个生日就将其埋葬。本文并没有计划太多的代码-可以在文档中找到编写应用程序和使用各个部分的示例(那里有很多代码;)。本文更多是概述。
谁需要它?
这个问题的答案有些自私-首先,当然是我自己。我在使用现有框架构建Web应用程序方面有一些经验,我经常引起自己的思考:“是的,一切都很酷,但是如果是这样……。这是一个商业广告...”...我们大多数人或多或少都会遇到这样的事实,即有些事情不喜欢并且希望(甚至不得不)改变它们。我试图将自己喜欢的工具整合在一起。我希望我并不孤单,并且希望有人会接近这些想法。 Crax背后的主要思想是,它不会施加任何特定的开发风格。例如,我们不需要名称空间,我们不想将逻辑划分为应用程序,我们想要快速部署两条路由并驱动请求和响应。好的,在这种情况下,我们可以只创建一个文件应用程序并得到我们想要的。但是相反的情况也是可能的,这也不是问题。 Crax倡导的第二件事是简单性。开始时需要最少的代码和最少的文档阅读。如果一个刚开始学习Python的人计划使用该框架,那么他应该能够轻松地克服入门的门槛。
如果您查看通过所有
TechEmpower测试所需的代码行数(请参见下文),那么由一个文件组成的应用程序中的Crax比其他所有参与者都更紧凑,并且没有“收缩”该文件的目的。真的没有什么可写的了。综上所述,我们可以说Crax适合于非常不同的任务范围和非常广泛的不同程度培训的程序员。
为什么不使用现有工具?
为什么不?此外,如果您确切地知道要使用哪个工具,最适合您当前任务的工具,那么您已经使用了该工具并知道所有细微差别。当然,您会选择自己所知道和适合的。没有(永远不会)将Crax定位为“%framework_name%killer”。不会有任何搅动类型:“紧急抛出%framework_name%,重写Crax上的所有内容,并立即注意到销售额的增加
首先,它足够快。它是使用ASGI接口编写的(请在此处阅读规范),并且比Flask或Django 1. *,2。*快得多。但是Crax当然不是唯一使用ASGI的Python框架,初步测试表明,它与使用该技术的其他框架有很好的竞争。为了进行比较,我们使用了TechEmpower性能评级测试,不幸的是,Crax与本轮中间添加的其他框架一样,只会进入下一个框架,然后您可以在图形问题中看到结果。但是,在每个请求请求之后,Travis运行测试,您可以在Travis日志中看到框架的比较特征。下面的链接是特拉维斯日志Python框架的长footcloth与按字母顺序排列的名字从A到F这里...您可以尝试读取日志并比较Crax,例如,与apidaora比较,结果会很好。该图下方是19轮测试中的当前状态。
当然,尽管如此,我们将只能在下一轮中看到真实的结果和真实的结果。
但是,如上所述,我们拥有不少快速且经过验证的工具。
同样的异步,对websockets和其他乐趣的本地支持。
假设是Starlette或FastApi。它们绝对是令人惊叹的框架,拥有对开发这些产品感兴趣的广大社区。值得注意的是,Crax在意识形态上与Starlette或FastAPI最相似,并且某些想法已
from crax.utils import get_settings_variable
base_url = get_settings_variable('BASE_URL')
但是,当配置文件开始因变量和设置过度增长而我们想要访问它们时,这似乎是一个可疑的优势,这变得很重要。
我要谈的下一个重要细节是应用程序结构的组织。当您有一个小项目时,可以将所有逻辑放在一个文件中,这是一回事。但是,当您编写更具全局性的内容时,您可能希望根据其逻辑将视图,模型,路线描述等分开。在这种情况下,会想到出色的Flask蓝图或Django应用程序。 Crax在这种意义上谈论名称空间。最初,您的应用程序旨在
主项目文件中包含的一组python软件包。顺便说一句,名称空间(应用程序的各个部分)可以递归嵌套(Hello Flask),它们中的文件名无关紧要。为什么这样 它给我们带来什么?
首先,路由。命名空间将根据命名空间的位置自动创建uri(但是可以控制)。例如:
from crax.urls import Route, Url, include
url_list = [
Route(Url('/'), Home),
Route(Url('/guest_book'), guest_view_coroutine),
include('second_app.urls'),
include('second_app.nested.urls'),
include('third_app.urls')
]
用斜杠替换点,您将获得uri到您的名称空间(当然,通过添加最终处理程序)。既然我们已经提到了路由,我们将对其进行更详细的介绍。
除了使用正则表达式的常规工作或通过Django路径进行的工作之外,Crax还提供了许多有趣的可能性。
# URL defined as regex with one floating (optional) parameter
Url(r"/cabinet/(?P<username>\w{0,30})/(?:(?P<optional>\w+))?", type="re_path")
# General way to define URL
Url("/v1/customer/<customer_id>/<discount_name>/")
但是,可以将多个Urls绑定到一个处理程序。
from crax.urls import Route, Url
class APIView(TemplateView):
template = "index.html"
urls = [
Route(
urls=(
Url("/"),
Url("/v1/customers"),
Url("/v1/discounts"),
Url("/v1/cart"),
Url("/v1/customer/<customer_id:int>"),
Url("/v1/discount/<discount_id:int>/<optional:str>/"),
),
handler=APIView)
]
您自己可以想到它对您有用的地方。并且,在“伪装”模式下存在解析器的操作模式。例如,您只想分发带有模板的某种目录,而不想要其他任何东西。也许这是Sphinx文档,或者类似的东西。您可以随时这样做:
import os
from crax.urls import Url, Route
class Docs(TemplateView):
template = 'index.html'
scope = os.listdir('docs/templates')
URL_PATTERNS = [
Route(urls=(
Url('/documentation', masquerade=True),
handler=Docs),
]
太好了,现在可以使用一个处理程序成功渲染docs / templates目录中的所有模板。一个好奇的读者会说这里根本不需要python,而所有这些都只能在条件Nginx的帮助下完成。我完全同意,直到有必要(例如,按角色或在一侧的某个位置分发这些模板)之前,不需要其他逻辑。
但是,回到我们的
Crax中没有ORM。这是不应该的。无论如何,直到SQLAlchemy提供异步解决方案。但是,已声明使用数据库(Postgres,MySQL和SQLite)。这意味着可以基于Crax BaseTable编写自己的模型。在后台,这是SQLAlchemy Core Table的非常薄的包装,它可以完成Core Table可以做的所有事情。对于可能需要的东西。也许要做类似的事情。
from crax.database.model import BaseTable
import sqlalchemy as sa
class BaseModelOne(BaseTable):
# This model just passes it's fields to the child
# Will not be created in database because the abstract is defined
parent_one = sa.Column(sa.String(length=50), nullable=False)
class Meta:
abstract = True
class BaseModelTwo(BaseTable):
# Also passes it's fields to the child
# Will be created in database
parent_two = sa.Column(sa.String(length=50), nullable=False)
class MyModel(BaseModelOne, BaseModelTwo):
name = sa.Column(sa.String(length=50), nullable=False)
print([y.name for x in MyModel.metadata.sorted_tables for y in x._columns])
# Let's check our fields ['name', 'id', 'parent_one', 'parent_two']
并且为了能够与迁移一起工作。 Crax迁移是SQLAlchemy Alembic之上的一些代码。因为我们在谈论名称空间和逻辑分离,所以
很显然,我们希望将迁移与该名称空间的其他逻辑存储在同一包中。这就是Crax迁移的工作方式。所有迁移将根据其命名空间进行分配,如果此命名空间暗示要与其他数据库一起使用,则在迁移目录内部将划分为相应数据库的目录。脱机迁移也是如此-所有* .sql文件将根据名称空间和模型数据库进行拆分。我不会在这里谈论编写查询的问题-它在文档中,我只会说您仍在使用SQLAlchemy Core。
同样,名称空间暗示了模板的方便存储(支持继承和其他Jinja2功能以及以现成的CSRF令牌或URL生成的形式提供的一些便利)。也就是说,所有模板都是结构化的。好吧,当然,我并不会陷入辉煌的2007年,我知道模板(即使它们是异步呈现的)在2020年将很少有需求。而且,很可能您很高兴将前端和后端的逻辑分开。 Crax在这方面做得很好,可以在Github上查看结果。
这里VueJs用作前端。并且由于我们拥有某种API,我们可能希望制作交互式文档。 Crax可以根据您的路由列表和处理程序文档字符串开箱即用地构建OpenAPI(Swagger)文档。当然,所有示例都在文档中。
在继续进行简要概述中最有趣的部分之前,有必要先讨论一下Crax已经提供了哪些有用的电池。
自然,调试模式是指可以在发生不幸的页面上的浏览器中直接读取错误和完整跟踪的情况。调试模式可以通过
内置记录器,可以同时写入指定的文件并将日志发送到控制台(或执行一件事)。能够分配您自己的记录器,而不是默认记录器。通过在配置中添加两行(如有必要,还可以自定义)来支持哨兵。
两种类型的预装中间件。第一个在应用程序处理请求之前被处理,第二个之后。
内置对CORS标头的支持。您只需要在配置中声明CORS规则。
能够直接在现场定义每个处理程序可用的方法。每个处理程序将使用指定的HTTP方法列表(+ HEAD和OPTIONS),或者仅使用GET,HEAD和OPTIONS。
能够指定此处理程序仅对授权用户可用,或仅对Administrators组中的用户可用,或仅对超级用户角色的成员可用。
有HMAC签名会话的授权,您无需进入数据库,并拥有许多用于创建和管理用户的工具。您可以启用授权后端支持并获得预设用户和许多可使用的工具。但是,像大多数Crax工具一样,您可以将其保留,使用和编写自己的工具。您不能使用授权,数据库,模型,迁移,视图并完全编写自己的自定义解决方案。您无需为此付出任何努力,没有启用它-并非如此。
有几种类型的Response和几种基于类的处理程序可以帮助您更快,更简洁地编写应用程序。在这种情况下,您自己的也可以使用,而不会继承自内置的。
from crax.views import BaseView
# Written your own stuff
class CustomView:
methods = ['GET', 'POST']
def __init__(self, request):
self.request = request
async def __call__(self, scope, receive, send):
if self.request.method == 'GET':
response = TextResponse(self.request, "Hello world")
await response(scope, receive, send)
elif self.request.method == 'POST':
response = JSONResponse(self.request, {"Hello": "world"})
await response(scope, receive, send)
# Crax based stuff
class CustomView(BaseView):
methods = ['GET', 'POST']
async def get(self):
response = TextResponse(self.request, "Hello world")
return response
async def post(self):
response = JSONResponse(self.request, {"Hello": "world"})
return response
class CustomersList(TemplateView):
template = 'second.html'
# No need return anything in case if it is TemplateView.
# Template will be rendered with params
async def get(self):
self.context['params'] = self.request.params
CSRF保护支持。生成令牌,检查请求正文中是否存在令牌,
禁用对特定处理程序的验证。
支持ClickJacking保护(框架,iframe,嵌入...呈现策略)
支持在应用程序开始处理请求之前检查请求的最大允许正文大小。
本机websocket支持。让我们以文档中的示例为例,并编写一个简单的应用程序,该应用程序可以按广播,每个用户组或特定用户的消息发送websocket消息。假设我们有组“男孩”和“女孩”(可以添加组“父母”)。我们可以为示例编写类似的内容(当然,这不是产品代码)。
#app.py
import asyncio
import json
import os
from base64 import b64decode
from functools import reduce
from crax.auth import login
from crax.auth.authentication import create_session_signer
from crax.auth.models import Group, UserGroup
from crax.response_types import JSONResponse
from crax.urls import Route, Url
from crax.views import TemplateView, WsView
from sqlalchemy import and_, select
from websockets import ConnectionClosedOK
BASE_URL = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SECRET_KEY = "SuperSecret"
MIDDLEWARE = [
"crax.auth.middleware.AuthMiddleware",
"crax.auth.middleware.SessionMiddleware",
]
APPLICATIONS = ["ws_app"]
CLIENTS = {'boys': [], 'girls': []}
class Home(TemplateView):
template = "index.html"
login_required = True
class Login(TemplateView):
template = "login.html"
methods = ["GET", "POST"]
async def post(self):
credentials = json.loads(self.request.post)
try:
await login(self.request, **credentials)
if hasattr(self.request.user, "first_name"):
context = {'success': f"Welcome back, {self.request.user.username}"}
status_code = 200
else:
context = {'error': f"User or password wrong"}
status_code = 401
except Exception as e:
context = {'error': str(e)}
status_code = 500
response = JSONResponse(self.request, context)
response.status_code = status_code
return response
class WebSocketsHome(WsView):
def __init__(self, request):
super(WebSocketsHome, self).__init__(request)
self.group_name = None
async def on_connect(self, scope, receive, send):
# This coroutine will be called every time a client connects.
# So at this point we can do some useful things when we find a new connection.
await super(WebSocketsHome, self).on_connect(scope, receive, send)
if self.request.user.username:
cookies = self.request.cookies
# In our example, we want to check a group and store the user in the desired location.
query = select([Group.c.name]).where(
and_(UserGroup.c.user_id == self.request.user.pk, Group.c.id == UserGroup.c.group_id)
)
group = await Group.query.fetch_one(query=query)
self.group_name = group['name']
# We also want to get the username from the user's session key for future access via direct messaging
exists = any(x for x in CLIENTS[self.group_name] if cookies['session_id'] in list(x)[0])
signer, max_age, _, _ = create_session_signer()
session_cookie = b64decode(cookies['session_id'])
user = signer.unsign(session_cookie, max_age=max_age)
user = user.decode("utf-8")
username = user.split(":")[0]
val = {f"{cookies['session_id']}:{cookies['ws_secret']}:{username}": receive.__self__}
# Since we have all the information we need, we can save the user
# The key will be session: ws_cookie: username and the value will be an instance of uvicorn.WebSocketProtocol
if not exists:
CLIENTS[self.group_name].append(val)
else:
# We should clean up our storage to prevent existence of the same clients.
# For example due to page reloading
[
CLIENTS[self.group_name].remove(x) for x in
CLIENTS[self.group_name] if cookies['session_id'] in list(x)[0]
]
CLIENTS[self.group_name].append(val)
async def on_disconnect(self, scope, receive, send):
# This coroutine will be called every time a client disconnects.
# So at this point we can do some useful things when we find a client disconnects.
# We remove the client from the storage
cookies = self.request.cookies
if self.group_name:
try:
[
CLIENTS[self.group_name].remove(x) for x in
CLIENTS[self.group_name] if cookies['session_id'] in list(x)[0]
]
except ValueError:
pass
async def on_receive(self, scope, receive, send):
# This coroutine will be called every time we receive a new incoming websocket message.
# Check the type of message received and send a response according to the message type.
if "text" in self.kwargs:
message = json.loads(self.kwargs["text"])
message_text = message["text"]
clients = []
if message["type"] == 'BroadCast':
clients = reduce(lambda x, y: x + y, CLIENTS.values())
elif message["type"] == 'Group':
clients = CLIENTS[message['group']]
elif message["type"] == 'Direct':
username = message["user_name"]
client_list = reduce(lambda x, y: x + y, CLIENTS.values())
clients = [client for client in client_list if username.lower() in list(client)[0]]
for client in clients:
if isinstance(client, dict):
client = list(client.values())[0]
try:
await client.send(message_text)
except (ConnectionClosedOK, asyncio.streams.IncompleteReadError):
await client.close()
clients.remove(client)
URL_PATTERNS = [Route(Url("/"), Home), Route(Url("/", scheme="websocket"), WebSocketsHome), Route(Url("/login"), Login)]
DATABASES = {
"default": {
"driver": "sqlite",
"name": f"/{BASE_URL}/ws_crax.sqlite",
},
}
app = Crax('ws_app.app')
if __name__ == "__main__":
if sys.argv:
from_shell(sys.argv, app.settings)
<!-- index.html -->
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Crax Websockets</title>
</head>
<body>
<div id="wsText"></div>
<form>
<input id="messageText"><br>
<select id="targetGroup">
<option>boys</option>
<option>girls</option>
</select>
<select id="messageType">
<option>BroadCast</option>
<option>Group</option>
<option>Direct</option>
</select>
<select id="userNames">
<option>Greg</option>
<option>Chuck</option>
<option>Mike</option>
<option>Amanda</option>
<option>Lisa</option>
<option>Anny</option>
</select>
</form>
<a href="#" id="sendWs">Send Message</a>
<script>
var wsText = document.getElementById("wsText")
var messageType = document.getElementById("messageType")
var messageText = document.getElementById("messageText")
var targetGroup = document.getElementById("targetGroup")
var userName = document.getElementById("userNames")
var sendButton = document.getElementById("sendWs")
ws = new WebSocket("ws://127.0.0.1:8000")
ws.onmessage = function(e){
wsText.innerHTML+=e.data
}
sendButton.addEventListener("click", function (e) {
e.preventDefault()
var message = {type: messageType.value, text: messageText.value}
var data
if (messageText.value !== "") {
if (messageType.value === "BroadCast"){
// send broadcast message
data = message
}
else if (messageType.value === "Group"){
// send message to group
data = Object.assign(message, {group: targetGroup.value})
}
else if (messageType.value === "Direct"){
// send message to certain user
data = Object.assign(message, {user_name: userName.value})
}
ws.send(JSON.stringify(data))
}
})
</script>
</body>
</html>
<!-- login.html -->
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Crax Websockets</title>
</head>
<body>
<form>
<input id="username">
<input id="password" type="password">
</form>
<div id="loginResults"></div>
<a href="#" id="sendLogin">Login</a>
<script>
var loginButton = document.getElementById("sendLogin")
var loginResults = document.getElementById("loginResults")
var username = document.getElementById("username")
var password = document.getElementById("password")
loginButton.addEventListener("click", function (e) {
e.preventDefault()
if (username.value !== "" && password.value !== "") {
var xhr = new XMLHttpRequest()
xhr.overrideMimeType("application/json")
xhr.open("POST", "/login")
xhr.send(JSON.stringify({username: username.value, password: password.value}))
xhr.onload = function () {
var result = JSON.parse(xhr.responseText)
if ("success" in result){
loginResults.innerHTML+="<h5 style='color: green'>"+result.success+ "</h5>"
}
else if ("error" in result) {
loginResults.innerHTML+="<h5 style='color: red'>"+result.error+ "</h5>"
}
}
}
})
</script>
</body>
</html>
完整的代码可以在Crax文档中查看。
好了,该是本文中最有趣的时候了。
为什么没有必要?
首先,如上所述,有几个相同的框架,并且已经形成了一个社区。 Crax是一个婴儿,一个星期大。单身军几乎可以保证该项目迟早会被放弃。令人遗憾的是,在桌面上工作,只为您自己和从Syktyvkar发行Vasily发布发行版和更新,这一事实比社区进行项目开发的时间长得多。同时,该项目没有2020年必须具备的许多功能。例如:不支持JWT(JOSE)。没有对OAuth2工具的现成支持。不支持GraphQL。很明显,您可以自己为项目编写此代码,但Starlette或FastAPI已经拥有它。我只需要写这个(是的,它在计划中)。最后的计划会有一些。
Netflix和Microsoft的开发人员撰写有关FastAPI的文章。关于Crax写道noname,不知道它出现在哪里,谁知道确切的能力在后天的深渊里发生。 他们
不会叫我的名字白痴一个蒸笼。
我的母亲哭到了晚上,因为她生下了一个怪物......
(C)
这是很重要的。这就是声誉和生态系统。 Crax也没有。没有这些重要的东西,该项目将被保证直接运往垃圾填埋场而不会诞生。
值得理解。上面写的不是在敲门课,而是在火车上无家可归的人的文字。这是一个清醒的评估,并警告“生产就绪解决方案”不仅是测试覆盖源代码的结果,而且是对项目中使用的技术,方法和解决方案的成熟度的一般评估。
如果您只是刚开始接触Python并尝试使用框架,那么您就处于危险之中:很可能,您不会在SO上找到该问题的答案,不幸的是,也许经验丰富的同志可能会没有帮助您。
目标
我计划要做的第一件事当然是添加一些必需的东西,例如JWT(JOSE),OAuth2和GraphQL支持。这将使我和感兴趣的人更轻松地工作。实际上,这是Crax的主要目标-使某人的工作容易一些。也许届时TechEmpower的新一轮比赛将开始,基准将变得更加明显。之后,甚至有可能对社区产生一些兴趣。
有一种基于Crax编写CMS的想法。
如果我没记错(如果我记错了,请更正),我们的工具包中还没有Python中的异步CMS。我可能会改变主意,决定写某种电子商务解决方案。但是,显然,为了防止Crax在到达浮标之前溺水,需要在其基础上做一些有趣的事情。也许发烧友对此会感兴趣。爱好者是免费的。因为这里没有钱,而且很可能不会。Crax对每个人都是完全免费的,而我没有得到这份工作的毛钱。因此,计划在“漫长的冬天的夜晚”进行开发,也许在来年,将会诞生一些有趣的东西。
结论
我当时在考虑将哪个文章包含在本文中(顺便说一下,这是我在该资源上的第一篇出版物)。甚至值得将其放在标签“ I'm PR”下。是什么让我改变了主意:首先,它没有任何广告特性。
没有呼叫“男孩,紧急注册拉动请求”。没有想法在这里找到赞助商。我什至没有想到我给您带来了您从未见过(当然是见过)的东西。您可以从我既是本文的作者,又是该工具的作者的思想中抽象出来的,并将其视为一篇评论文章。而且,是的,这是最好的方法。如果您牢记确实如此,那对我来说将是一个极好的结果。
这也许就是全部。
“所以……现在是时候钓竿了。
-为什么?
-哈里斯的红帽子吓到了所有的鱼。
(c)
GitHub文档上的代码