Python-FastAPI based on asyncio asynchronous ecological development asynchronous blog (2) communication logic articles

Posted Jun 16, 20208 min read

Project address

Blog address

The first version of Frodo has been implemented. Before the next version, I will organize the current development ideas into three articles, namely data articles, communication articles, and asynchronous articles.

This article comes to the logical process of realizing specific functions. In the web application summary, I personally prefer to make the business process "communication". Because the whole process is to send the data organization and processing to the front end in the background, the protocol of this process can be different(http(s), websocket), the method may be different(rcp, ajax, mq), and the returned content format is different(json, xml, html(templates), the early years of Flash, etc.); I just talked about front-end and back-end communication. In fact, communication problems are involved between logical modules, processes, and even subsequent containers. This article first introduces the core of Web communication, front-end and back-end communication.

Separation of template technology from front and back ends

  • Template technology:Web technology widely used in the first ten years of this century, he has a more famous name of MVC mode. The core idea is to use backend code to write data in the HTML template, and the template engine will return the rendered HTML. Django embeds this technology, other Python frameworks need to rely on separate templates such as jinjia, Mako, etc. Other languages such as Java's JSP also use this model. His characteristic is that the operation is direct, and the corresponding data is written directly where needed. You can also use the back-end language to write logic on the page directly, and the development speed is fast. But the shortcomings are also obvious. The front-end and back-end are heavily coupled and difficult to maintain, which is not suitable for large-scale projects.

    • Protocol:http
    • Method:Both
    • Content:html(templates)
  • Front-end and back-end separation:The current mainstream model, when the project is getting bigger and bigger, the demand for front-end engineering has spawned the webpack tool. Then the Vue, React, and Angular frameworks focus on the MVVC pattern, that is, only get data from the backend, rendering and business logic into the front-end framework. In this way, front-end developers can be separated to the greatest extent.

    • Agreement:all
    • Method:Both
    • Content:json/xml

Mako template and his friend FastAPI-Mako

Frodo uses templates as the front desk of the blog display, considering that there are few pages in this part, the logic is simple, and the backend staff is easy to maintain. The template is completely sufficient. _There is no outdated technology, only suitable technology_.

Mako is one of the mainstream python templates. His native interface can actually be used directly, but some repetitive logic requires us to pack it:

  • Several context variables fixed in the template

    • Request object(the request object used by the back-end framework, which exists in Flask, Django, and fastapi), the template needs to use some of his methods and attributes, such as reverse addressing request.url_for(),, or even the content in request.Session
    • Request context(mainly refers to the body, friends who have contacted with Web development can list the main request body:Formdata, QueryParam, PathParam, which may be used in the template)
    • Return to context(without encapsulation provided by Ye Tao)
  • Automatic addressing of template files

  • Static file addressing

  • Template exception handling

Like Flask, the routing of fastapi is also functional. In order to encapsulate the above template functions into routing functions, the direct approach is to use python decorators. Eventually up to the following effects:

from fastapi import APIRouter, Request, HTTPException
from fastapi.responses import HTMLResponse
from models import cache, Post
from ext import mako

@router.get('/archives', name='archives', response_class=HTMLResponse)
@mako.template('archives.html') # Specify the template file name
async def archives(request:Request):# Need to pass Request explicitly
    post_data = await Post.async_filter(status=Post.STATUS_ONLINE)
    post_obj = [Post(**p) for p in post_data]
    post_obj = sorted(post_obj, key=lambda p:p.created_at, reverse=True)
    rv = dict()
    for year, items in groupby(post_obj, lambda x:x.created_at.year):
        if year in rv:rv[year].extend(list(items))
        else:rv[year]= list(items)
    archives = sorted(rv.items(), key=lambda x:x[0], reverse=True)
    # Only return context
    return {'archives':archives}

In fact, it s easy to understand, the only thing to explain is why you should explicitly pass request, fastapi avoids passing request to the greatest extent, this is the same as Flask, you can do it with the local thread stack To distinguish the context of different requests. But the template needs to be frequently reversed, similar to:

%for year, posts in archives:

${ year}


@mako Simple decorator is complete in LouisYZK/FastAPI-Mako, interested friends can check it out.

In addition, there is the @cached decorator, which caches the return result template of the function. If the data of the current page does not change, the next visit will directly get the data from redis. The detailed logic will be introduced in the CRUD logic below.

Communication logic of CRUD

This section is about all data models. Individual data storage methods such as Posts and Activity have multiple ways, and they need more tricks. All data operations follow the following process:

The "DataModel" in the control use case is the data class designed in the data chapter. They have several methods to deal with the needs of CRUD, the most important of which are two:

  • Generate operation SQL
  • Generate the key used by KV database cache

Both of the above points use some tricks of sqlalchemy and python decorator. You can focus on the source code models/ and models/

It is worth mentioning that the implementation of update and delete cache:

## clear_mc method in
async def clear_mc(*keys):
    redis = await get_redis()
    print(f'Clear cached:{keys}')
    assert redis is not None
    await asyncio.gather(*[redis.delete(k) for k in keys],

## base class __flush__ method
from import clear_mc
async def __flush__(cls, target):
    await asyncio.gather(
        target.clear_mc(), return_exceptions=True

## target is a specific data instance, they rewrite the clear_mc() method to delete the specified different key, such as the rewriting of the following Post class:
async def clear_mc(self):
    keys = [
        MC_KEY_TAGS, MC_KEY_RELATED%(, 4),
    for i in [True, False]:
    for tag in await self.tags:
    await clear_mc(*keys)

This ensures that every time you create, update, or delete data, you can delete the relevant cache to maintain data consistency. You may have noticed that the operation to delete the cache is awaitable, which means that asynchronous can take advantage of concurrency. So we saw the use of asyncio.gather(*coros), he can delete multiple keys concurrently, because redis creates a connection pool, so that no multi-threading is used, and this is how asyncio implements io concurrency.(In fact, this point should be introduced in the asynchronous article, but this point is very important).


The requirements for certification come from two:

  • The background of the content management system can only be operated by the blog owner, such as blog posting and password modification.
  • Visitor comments require identity verification.

Authentication of the administrator--using JWT

JWT is one of the widely used authentication methods at present, and his advantages over cookies can refer to related articles. And fastapi has built-in support for JWT, we use him to verify it is very convenient.

Before talking about the specific implementation, you still have to think about his communication logic:

The above process represents the logic of login and the general logic of accessing the authentication API. Did you find the problem? Where is Token stored?

Where does Token exist? The server generates the Token client to receive, the next request will bring him. This kind of frequently used and small volume of data is most suitable for storage directly in memory. It is necessary to share global variables in the programming language, for example, multiprocess.Value is to solve this problem. But asynchronous is studied for the event loop, there is no concept of threaded process, at this time contextvar is specifically to solve the problem of asynchronous variable sharing, requires python greater than 3.7

fastapi helps us maintain this Token, it only needs to be defined as follows:

from import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/auth')

It means that the token generation path is /auth, and the oauth2_scheme shape is used as the source of the global dependent token. Whenever the interface needs to use the token, it only needs to:

async def list_users(token:str = Depends(oauth2_scheme)) -> schemas.CommonResponse:
    users:list = await User.async_all()
    users = [schemas.User(**u) for u in users]
    return {'items':users,'total':len(users)}

Depends is a feature of fastapi, written directly in the parameters of the interface function, you can execute some logic before the request, similar to middleware. The logic here is to check whether the request header carries Auth:Bear+Token, and if not, you cannot issue this request.

The token generation logic is completed on the login interface, which is almost the most complicated logic of Frodo:'/auth')
async def login(req:Request, username:str=Form(...), password:str=Form(...)):
    user_auth:schemas.User = \
            await user.authenticate_user(username, password)
    if not user_auth:
        raise HTTPException(status_code=400,
                            detail='Incorrect User Auth.')
    access_token_expires = timedelta(
    access_token = await user.create_access_token(
    return {

Basically follow the timing diagram in this section.

Access authentication-use Session

There are many access certifications, which are limited by the content of the blog. Anyone who visits Frodo should have Github, so he uses his authentication, the logic is as follows:

The whole logic is very simple, follow the Github authentication logic, another way such as WeChat scan code will have to change a set. Pay attention to the URL of the jump. At the same time, storing the visitor's information does not use JWT, because there is no limit to obsolescence, etc., the session cookie is the most direct.

async def oauth(request:Request):
    if'error' in str(request.url):
        raise HTTPException(status_code=400)
    client = GithubClient()
    rv = await client.get_access_token(code=request.query_params.get('code'))
    token = rv.get('access_token','')
        user_info = await client.user_info(token)
        return RedirectResponse(config.OAUTH_REDIRECT_PATH)
    rv = await create_github_user(user_info)
    ## Use session storage
    request.session['github_user']= rv
    return RedirectResponse(request.session.get('post_url'))

Note that fastapi needs to add middleware to open the session

from starlette.middleware.sessions import SessionMiddleware

app.add_middleware(SessionMiddleware, secret_key='YOUR KEY')

What is starlette? Isn't it fastapi? Simply speaking, the relationship between starlette and fastapi is the same as the relationship between werkzurg and Flask. The difference between WSGI and ASGI. Now the idea of ASGI is to go beyond WSGI. Of course, you must also set up a set of basic standards and tool libraries. .

Fine, communication logic is basically this, Frodo uses few communication models. The next "asynchronous article" is related to communication and data. The difference between asynchronous blogs and blogs implemented by general python is here.