Squall, API framework which looks ahead
Initially, it was a library for ASGI frameworks for publishing RBAC routing information to the MTAG API-Gateway. After some research, we have decided that this is the most expensive way and made a decision to create a framework which will deliver the best experience in the development of applications behind the API-Gateway.
Eventually, Squall is a part of the e2e solution for modern high-performance stacks.
1Kb no schema
30Kb no schema
1Kb schema
30Kb schema
More results and benchmark methodology here
Squall following own MTAG/Squall ASAP pattern. The idea of the ASAP pattern is pretty simple to understand. If you have all necessaries to do something you can do in the next steps, you should do it now.
Be careful. This pattern is mind-changing.
The batch operation is always better than a lot of small ones.
pip3 install python-squall
You also need some ASGI server. Let's install Uvicorn, the most popular one.
pip3 install uvicorn
Create example.py
with the following content
from typing import List, Optional
from dataclasses import dataclass
from squall import Squall
app = Squall()
@dataclass
class Item:
name: str
value: Optional[int] = None
@app.get("/get", response_model=List[Item])
async def handle_get() -> List[Item]:
return [
Item(name="null_value"),
Item(name="int_value", value=8)
]
@app.post("/post", response_model=Item)
async def handle_post(data: Item) -> Item:
return data
And run it
uvicorn example:app
Now, we are able to surf our GET endpoint on: http://127.0.0.1:8000/get
And let's play with curl
on POST endpoint
# curl -X 'POST' 'http://127.0.0.1:8000/post' -H 'Content-Type: application/json' -d '{"name": "string", "value": 234}'
{
"name": "string",
"value": 234
}
Type checking and validation is done by apischema for both, Request and Response.
# curl -X 'POST' 'http://127.0.0.1:8000/post' -H 'Content-Type: application/json' -d '{"name": "string", "value": "not_an_int"}'
{
"details": [
{
"loc": [
"value"
],
"msg": "expected type integer, found string"
},
{
"loc": [
"value"
],
"msg": "expected type null, found string"
}
]
}
OpenAPI for your app generates automatically based on route parameters and schema you have defined.
There are support for ReDoc and Swagger out of the box. You can reach it locally once your application started:
Squall provides familiar decorators for any method route registration on both, application itself and on nested routers.
Method | app | router * |
---|---|---|
GET | @app.get | @router.get |
PUT | @app.put | @router.put |
POST | @app.post | @router.post |
DELETE | @app.delete | @router.delete |
OPTIONS | @app.options | @router.options |
HEAD | @app.head | @router.head |
PATCH | @app.patch | @router.patch |
TRACE | @app.trace | @router.trace |
* router = squall.Router()
Nested routers supports prefixes and further nesting.
from squall import Router, Squall
animals_router = Router(prefix="/animals")
@animals_router.get("/")
async def get_animals():
return []
@animals_router.get("/cat")
async def get_cat():
return []
dogs_router = Router(prefix="/dogs")
@dogs_router.get("/list")
async def get_all_dogs():
return []
animals_router.include_router(dogs_router)
app = Squall()
app.include_router(animals_router)
Will give us
Nested routing is usually used for splitting applications into files and achieving better project structure.
Squall provides built-in blazing-fast compression based on Intel® Intelligent Storage Acceleration Library (Intel® ISA-L) using awesome Python's isal library as binding.
Compared to Python's builtins ISA-L can deliver up to 20 times faster compression. Such in-app performance does game-changing opportunities for the entire system set up,
In order to enable compression you have to path compression config to Squall app
from squall import Squall
from squall.compression import Compression
app = Squall(compression=Compression())
For more details check compression settings
Accept-Encoding header also required. Squall supports gzip, deflate options for it.
There are four kinds of parameters that developers can get from HTTP headers. Squall offers an interface for their conversion and validation.
"Path" is a dynamic value specified by developers in the route URL.
from squall import Squall, Path
app = Squall()
@app.get("/company/{company_id}/employee/{employee_id}")
async def get_company_employee(company_id: int, employee_id = Path()):
return {
"company_id": company_id,
"employee_id": employee_id,
}
Squall determinate affiliation of the variable with path by any of following ways:
Path
instance Specifics:
str
, bytes
, int
, float
Union
, Optional
, not allowed. Because a path can't have an undefined value. Also, parameters must have a strong conversion contract.str
Shares common configuration contract for head entities. Please, read more here.
"Query" is a way get query string parameters value(s).
from typing import List
from squall import Squall, Query
app = Squall()
@app.get("/")
async def get_company_employee(company_id: int = Query(), employee_ids: List[int] = Query()):
return {
"company_id": company_id,
"employee_ids": employee_ids,
}
Specifics:
str
, bytes
, int
, float
, Optional
, List
"Header" is a way to get header value(s). Shares common behavior with Query
from typing import List
from squall import Squall, Header
app = Squall()
@app.get("/")
async def get_company_employee(company_id: int = Header(), employee_ids: List[int] = Header()):
return {
"company_id": company_id,
"employee_ids": employee_ids,
}
Specifics:
str
, bytes
, int
, float
, Optional
, List
"Cookie" is a way get cookie value.
from typing import List
from squall import Squall, Cookie
app = Squall()
@app.get("/")
async def get_company_employee(user_id: int = Cookie()):
return {
"user_id": user_id,
}
Specifics:
str
, bytes
, int
, float
, Optional
All head fields share common configuration pattern which include the following list of parameters:
default
, default value to assignalias
, replaces source key where to get the value fromtitle
, title for schema specificationdescription
, description for schema specificationvalid
, instance of validator, squall.Num
or squall.Str
example
, example for schema specificationexamples
, multiple examples for schema specificationdeprecated
, mark parameter as deprecated, will appear in specificationAt the moment, Squall provides following validators that developer can apply to HEAD parameters values:
squall.Num
- int
, float
validator. Following conditions are supported: gt
, ge
, lt
, le
squall.Str
- str
, bytes
validator. Following conditions are supported: min_len
, max_len
Please, take a look at the related test suite
Schema defined using dataclasses behind the scene validated by awesome apischema. Please follow their documentation for build validation.
There are things strictly important to remember:
If response_model is equal to the handler return annotation Squall expects exactly these types and will not perform mutations to dataclasses, etc. Type checking will be done during serialization.
Handy to save some resources working with ORM. For instance SQL Alchemy dataclass mapping
from typing import List, Optional
from dataclasses import dataclass
from squall import Squall
app = Squall()
@dataclass
class Item:
name: str
value: Optional[int] = None
@app.get("/get", response_model=List[Item])
async def handle_get() -> List[Item]:
return [
Item(name="null_value"),
Item(name="int_value", value=8)
]
The following example demonstrates a different scenario. Where response expects to receive from handler Python primitives and Sequences/Maps only. With this scenario, all response data will be processed through the filling of the relevant model.
from typing import List, Optional
from dataclasses import dataclass
from squall import Squall
app = Squall()
@dataclass
class Item:
name: str
value: Optional[int] = None
@app.get("/get", response_model=List[Item])
async def handle_get():
return [
{"name": "null_value"},
{"name": "int_value", "value": 8}
]
To trace internal actions next packages must be installed:
pip install opentelemetry-api opentelemetry-sdk
Having installed libs initial application and OpenTelemetry configuration should be performed as shown bellow:
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
BatchSpanProcessor,
ConsoleSpanExporter,
)
from squall import Squall
trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(ConsoleSpanExporter()))
app = Squall(trace_internals=True)
@app.get("/get")
async def handle_get() -> dict:
return {"Hello": "World"}
For detailed config have a look at Opentelemetry docs
Many thanks to @tiangolo and the entire FastAPI community. Squall development started from hard-forking this superior developers-friendly framework.
0.1.x
- Initial project publication
0.2.x
- Intel® ISA-L based compression
0.3.x
- Observability based on OpenTelemetry with switchable Squall internals tracing.
0.4.x
- Dependency Injector integration
0.5.x
- YARL and aio-MultiDict integration
0.6.x
- Fine-tuning for __slots__
, LEGB, attribute access.
0.7.x
- MTAG integration
0.8.x
- Starts new SGI initiative
License: MIT
Faster zlib and gzip compatible compression and decompression by providing python bindings for the ISA-L library.
License: MIT
JSON (de)serialization, GraphQL and JSON schema generation using Python typing.
apischema makes your life easier when dealing with API data.
License: MIT or Apache 2.0
orjson is a fast, correct JSON library for Python. It benchmarks as the fastest Python library for JSON and is more correct than the standard json library or other third-party libraries. It serializes dataclass, datetime, numpy, and UUID instances natively.
License: BSD 3
Starlette is a lightweight ASGI framework/toolkit, which is ideal for building high performance async services.
Squall follows the next versioning contract:
AA.BB.CC
AA
- Major changes, backward compatibility breaksBB
- Minor changes, new featuresCC
- Patch, bug fixesMIT