[CleanArchitecture with Python] Part2: Frameworks & Drivers Layer: Introducing the Web

In the previous Part 1, we made it as monolithic as possible.

  1. ** Receive a POST request and save a note **

  2. ** Receive a GET request and refer to the saved memo **

We have prepared a memo API just for you.

In this article, the explanation is based on the following code created in the previous chapter.

Part1 : https://qiita.com/y_tom/items/ac6f6a08bdc374336dc4

1. Receive a specification change request for the deliverable

I received a request to change the specifications of the "API created using the Flask framework" created in Part1.

** "Let's adopt FastAPI instead of Flask for web application framework." **

In Part 1, let's consider a design that is resistant to specification changes, assuming this specification change request.


I haven't encountered many cases where I want to replace the framework, but I thought it would be an easy-to-understand case as an introduction, so I adopted it.

As an aside, this is my most recent experience, but due to changes in market conditions, the Response Header of a certain Web application suddenly became available. There was a situation where I wanted to give a specific Header.

However, since the Header attribute was added in recent years, the Web application framework adopted at that time was In some cases, it didn't support the Header attribute and was forced to change the web application framework itself. (In the end, I wrote the Header raw in the custom header and responded, and I got nothing, but ...)

2. Concerns when responding to specification change requests with the current design

Now, let's get back to the story.

Currently, the following processes are collectively described in main.py.

  1. Accept requests through the framework
  2. Execute the process originally expected for the application (acquire and save memo)

main.py : https://github.com/y-tomimoto/CleanArchitecture/blob/master/part1/app/main.py

Coding when making changes in the current design

What happens when you change the framework you use in your current design?

If you try to change the framework from Flask to Fast API You will probably make the following modifications to the existing main.py.

  1. Rewrite the router configured by the framework
  2. Rewrite the response format
  3. Rewrite the error handler
  4. Rewrite how to start the app

If you keep the current design and make actual modifications to the existing main.py, you will see something like the following.

main.py

from http import HTTPStatus
- from flask import Flask, request, jsonify, make_response
+ from fastapi import FastAPI, Form, Response
+ import uvicorn
from mysql import connector

- app = Flask(__name__) 
+ app = FastAPI()

#Settings for DB connection
config = {
    ...
}

def exist(memo_id: int) -> bool:
    ...


- @app.route('/memo/<int:memo_id>')
+ @app.get('/memo/{memo_id}') 
def get(memo_id: int) -> str:

    ...

    
-   return jsonify(
-       {
-           "message": f'memo : [{result[1]}]'
-       }
-   )

+   return JSONResponse(
+       content={"message": f'memo : [{result[1]}]'
+   )


- @app.route('/memo/<int:memo_id>', methods=['POST'])
+ @app.post('/memo/{memo_id}')
- def post(memo_id: int) -> str:
+ async def post(memo_id: int, memo: str = Form(...)) -> str:


    ...

    
-   return jsonify(
-       {
-            "message": "saved."
-       }
-   )

+   return JSONResponse(
+      content={"message": "saved."}
+   )

- @app.errorhandler(NotFound)
- def handle_404(err):
-     json = jsonify(
-         {
-             "message": err.description
-         }
-     )
-     return make_response(json, HTTPStatus.NOT_FOUND)


+ @app.exception_handler(NotFound)
+ async def handle_404(request: Request, exc: NotFound):
+   return JSONResponse(
+       status_code=HTTPStatus.NOT_FOUND,
+       content={"message": exc.description},
+   )

- @app.errorhandler(Conflict)
- def handle_409(err):
-     json = jsonify(
-         {
-             "message": err.description
-         }
-     )
-     return make_response(json, HTTPStatus.CONFLICT)


+ @app.exception_handler(Conflict)
+ async def handle_409(request: Request, exc: Conflict):
+   return JSONResponse(
+       status_code=HTTPStatus.CONFLICT,
+       content={"message": exc.description},
+   )



if __name__ == '__main__':
-   app.run(debug=True, host='0.0.0.0') # DELETE
+   uvicorn.run(app=fastapi_app, host="0.0.0.0", port=5000) # NEW

Although it is possible to change the specifications by force in this way, there are some concerns.

Coding concerns when making changes in the current design

This fix modifies the framework code in main.py.

However, in main.py, not only the code related to the framework, but also the ** process of retrieving and saving notes **, which is originally expected of the application, is described.

Single Responsibility Principle: Principle of single responsibility: https://note.com/erukiti/n/n67b323d1f7c5

At this time, you may accidentally make unnecessary changes to the "process of acquiring and saving memos" that you originally expected from the application **.

I would like to avoid the situation where I make corrections while thinking that it may cause a problem by mistake in the code that is already working.

In this example, there are only two endpoints, but if this is a large service and you have multiple endpoints, this concern will be even greater.

Open / closed principle: Open / closed principle: https://medium.com/eureka-engineering/go-open-closed-principle-977f1b5d3db0

3. In response to the request, think about what kind of design was able to change the specifications smoothly based on Clean Architecture.

i. Reorganize design concerns

Concerns: ** May make unnecessary changes to existing code that is working properly **

Ii. What kind of design was it possible to avoid concerns and change the specifications?

This concern is due to the fact that main.py contains not only the framework but also the ** process of acquiring and saving notes ** that is originally expected of the application.

Therefore, the concern this time is main.py,

It seems to be solved by dividing it into ** framework ** and ** processing that is originally expected from the application **.

If the code is designed to be divided into roles, it seems that the scope of the modification can be limited to that role.

Iii. When the ideal design is interpreted by CleanArchitecture

In main.py,

  1. Receive requests with the flask framework
  2. Save the memo or get the memo

There are two processes.

In other words, in terms of CleanArchitecture,

  1. Web application framework
  2. Functions originally expected from the application

is.

In interpreting with CleanArchitecture, in the figure below,

  1. It seems that 1 can be described as the Web (part of the Frameworks & Drivers layer).

  2. Regarding 2, since it is a function that is originally expected from the application, it seems that it corresponds to either the Application Business Rules layer or the Enterprise Business Rules layer, but here, save the memo or get the memo Let's describe the function as MemoHandler.

It seems to be expressed as.

https___qiita-image-store.s3.amazonaws.com_0_293368_7ce1fb10-504e-16e0-8930-278b8a7f942d.jpeg

Now let's split main.py into the Frameworks & Drivers tier: Web and MemoHandler.

Iv. Actual coding

From main.py, call the Frameworks & Drivers layer: Web router, Design to call memo_handler.py from each router.

With this design, if you want to change the framework, just change the framework called in main.py. It does not modify the existing process memo_handler.py itself, so the existing process is not accidentally modified.

.
├── memo_handler.py 
└── frameworks_and_drivers
    └── web
        ├── fastapi_router.py
        └── flask_router.py

Frameworks & Drivers layer

frameworks_and_drivers/web/fastapi_router.py

from fastapi import FastAPI, Form, Request
from fastapi.responses import JSONResponse
from werkzeug.exceptions import Conflict, NotFound
from memo_handler import MemoHandler
from http import HTTPStatus

app = FastAPI()


@app.get('/memo/{memo_id}')
def get(memo_id: int) -> str:
    return JSONResponse(
        content={"message": MemoHandler().get(memo_id)}
    )


@app.post('/memo/{memo_id}')
async def post(memo_id: int, memo: str = Form(...)) -> str:
    return JSONResponse(
        content={"message": MemoHandler().save(memo_id, memo)}
    )


@app.exception_handler(NotFound)
async def handle_404(request: Request, exc: NotFound):
    return JSONResponse(
        status_code=HTTPStatus.NOT_FOUND,
        content={"message": exc.description},
    )


@app.exception_handler(Conflict)
async def handle_409(request: Request, exc: Conflict):
    return JSONResponse(
        status_code=HTTPStatus.CONFLICT,
        content={"message": exc.description},
    )


frameworks_and_drivers/web/flask_router.py


from flask import Flask, request , jsonify , make_response
from werkzeug.exceptions import Conflict,NotFound
from http import HTTPStatus
from memo_handler import MemoHandler
app = Flask(__name__)


@app.route('/memo/<int:memo_id>')
def get(memo_id: int) -> str:
    return jsonify(
        {
            "message": MemoHandler().get(memo_id)
        }
    )


@app.route('/memo/<int:memo_id>', methods=['POST'])
def post(memo_id: int) -> str:
    memo: str = request.form["memo"]
    return jsonify(
        {
            "message": MemoHandler().save(memo_id, memo)
        }
    )


@app.errorhandler(NotFound)
def handle_404(err):
    json = jsonify(
        {
            "message": err.description
        }
    )
    return make_response(json,HTTPStatus.NOT_FOUND)


@app.errorhandler(Conflict)
def handle_409(err):
    json = jsonify(
        {
            "message": err.description
        }
    )
    return make_response(json, HTTPStatus.CONFLICT)


MemoHandler

memo_handler.py

from mysql import connector
from werkzeug.exceptions import Conflict, NotFound

#config for sql client
config = {
    'user': 'root',
    'password': 'password',
    'host': 'mysql',
    'database': 'test_database',
    'autocommit': True
}


class MemoHandler:

    def exist(self, memo_id: int):
        #Create a DB client
        conn = connector.connect(**config)
        cursor = conn.cursor()

        # memo_Check if there is an id
        query = "SELECT EXISTS(SELECT * FROM test_table WHERE memo_id = %s)"
        cursor.execute(query, [memo_id])
        result: tuple = cursor.fetchone()

        #Close the DB client
        cursor.close()
        conn.close()

        #Check for existence by checking if there is one search result
        if result[0] == 1:
            return True
        else:
            return False

    def get(self, memo_id: int):

        #Check if there is a specified id
        is_exist: bool = self.exist(memo_id)

        if not is_exist:
            raise NotFound(f'memo_id [{memo_id}] is not registered yet.')

        #Create a DB client
        conn = connector.connect(**config)
        cursor = conn.cursor()
        # memo_Perform a search by id
        query = "SELECT * FROM test_table WHERE memo_id = %s"
        cursor.execute(query, [memo_id])
        result: tuple = cursor.fetchone()

        #Close the DB client
        cursor.close()
        conn.close()

        return f'memo : [{result[1]}]'

    def save(self, memo_id: int, memo: str):

        #Check if there is a specified id
        is_exist: bool = self.exist(memo_id)

        if is_exist:
            raise Conflict(f'memo_id [{memo_id}] is already registered.')

        #Create a DB client
        conn = connector.connect(**config)
        cursor = conn.cursor()

        #Save memo
        query = "INSERT INTO test_table (memo_id, memo) VALUES (%s, %s)"
        cursor.execute(query, (memo_id, memo))

        #Close the DB client
        cursor.close()
        conn.close()

        return "saved."



main.py

Switch the framework to be adopted on main.py.

main.py


import uvicorn
from frameworks_and_drivers.flask_router import app as fastapi_app
from frameworks_and_drivers.flask_router import app as flask_app

---

#When adopting flask as a framework
flask_app.run(debug=True, host='0.0.0.0')

---

#Fast as a framework_When adopting api
uvicorn.run(app=fastapi_app, host="0.0.0.0",port=5000)

4. What kind of specification changes have the design changes made it possible to withstand?

The final code is here. : https://github.com/y-tomimoto/CleanArchitecture/blob/master/part2

By cutting out each framework to Frameworks & Drivers layer: Web, and cutting out the processing originally expected from the application to MemoHandler, By simply calling the router you want to adopt with main.py, you can flexibly change the framework ** without modifying memo_handler.py, which is the process you originally expected from your application. I did.

https___qiita-image-store.s3.amazonaws.com_0_293368_7ce1fb10-504e-16e0-8930-278b8a7f942d.jpeg

This design implements one of the rules of CleanArchitecture, ** framework independence **.

Clean Architecture (Translated by The Clean Architecture): https://blog.tai2.net/the_clean_architecture.html

Framework Independence: The architecture does not rely on the availability of a full-featured library of software. This allows such frameworks to be used as tools and does not force the system to be forced into the limited constraints of the framework.

Recommended Posts

[CleanArchitecture with Python] Part2: Frameworks & Drivers Layer: Introducing the Web
[Part.2] Crawling with Python! Click the web page to move!
Download files on the web with Python
Try using the Python web framework Tornado Part 1
Try using the Python web framework Tornado Part 2
Make a breakpoint on the c layer with python
Web application made with Python3.4 + Django (Part.1 Environment construction)
Image processing with Python (Part 2)
Studying Python with freeCodeCamp part1
Bordering images with python Part 1
Web scraping with python + JupyterLab
Scraping with Selenium + Python Part 1
Studying Python with freeCodeCamp part2
Image processing with Python (Part 1)
Web API with Python + Falcon
Comparison of 4 Python web frameworks
Solving Sudoku with Python (Part 2)
Image processing with Python (Part 3)
Scraping with Selenium + Python Part 2
Call the API with python3.
Web application with Python + Flask ② ③
Save images on the web to Drive with Python (Colab)
Web scraping beginner with python
Streamline web search with python
Web application with Python + Flask ④
The first artificial intelligence. Challenge web output with python. ~ Flask introduction
[python, ruby] fetch the contents of a web page with selenium-webdriver
Visualize your pocket money files with the Python web framework Dash
Playing handwritten numbers with python Part 1
Extract the xz file with python
[Automation with python! ] Part 1: Setting file
Getting Started with Python Web Applications
Web scraping with Python First step
I tried web scraping with python.
Monitor Python web apps with Prometheus
Get the weather with Python requests
Get web screen capture with python
Find the Levenshtein Distance with python
Hit the Etherpad-lite API with Python
Install the Python plugin with Netbeans 8.0.2
Hit the web API in Python
I liked the tweet with python. ..
Master the type with Python [Python 3.9 compatible]
Automate simple tasks with Python Part0
[Automation with python! ] Part 2: File operation
Excel aggregation with Python pandas Part 1
How to crop the lower right part of the image with Python OpenCV
Hit a method of a class instance with the Python Bottle Web API
I tried using "Streamlit" which can do the Web only with Python
Make the Python console covered with UNKO
Introducing the BOT framework Minette for Python
[Python] Set the graph range with matplotlib
WEB scraping with Python (for personal notes)
Play handwritten numbers with python Part 2 (identify)
Process Pubmed .xml data with python [Part 2]
Behind the flyer: Using Docker with Python
Automate simple tasks with Python Part1 Scraping
Getting Started with Python Web Scraping Practice
Check the existence of the file with python
[Python] Get the variable name with str
[Python] Round up with just the operator