Performance API Design Guide

Use FastAPI

Recommend RESTful API Design

• https://restfulapi.net/
• Management platform must follow RESTful

File Structure

Principle

Have Startup function: Initialize Database connector, Global parameters, Common Class.

@router.on_event('startup')
async def startup():
    print(‘initializing…’)

MySQL: Use connection pool

from sqlalchemy import create_engine

class DataConnector:
    def __init__(self, db_url=None, pool_size=30) -> None:
        self.db_url = db_url
        self._engine = create_engine(
            db_url, pool_size=pool_size, pool_recycle=3600, pool_pre_ping=True
        )

    def clean_pool(self) -> bool:
        self._engine.dispose()
        return True

    def get_data(self, sql_string: str) -> list:
        try:
            conn = self._engine.connect()
            result = conn.execute(sql_string).fetchall()
            conn.close()
            return result
        except BaseException:
            self.clean_pool()
            raise

    def query_dict_list(self, sql_string: str, list_column: list) -> list:
        result = self.get_data(sql_string=sql_string)

        data = []
        for r in result:py
            data.append(
                {k: eval(v) if k in list_column else v for k,
                 v in r._asdict().items()}
            )

        return data

API Response Data: Use Base Model (pydantic)

from pydantic import BaseModel, Field

class Member(BaseModel):
    member_id: str = Field(title='memeber unique id',
                           example='4b14c168-825b-42c0-90be-9c6812aa27b7')
    installation_id: str = Field(title='device unique id', example='')
    app_id: str = Field(title='platform unique id',
                        example='b5b046d5175f160173887dc5df44c3a5')
    channel_id: str = Field(
        title='channel id', description='home page=index', example='index')

Error Handle: When processing raises errors, collect all failures to respond to errors.

# project_logger.py
from functools import wraps
from fastapi.responses import JSONResponse
import logging

proj_logger = logging.getLogger(__name__)
proj_logger.addHandler(logging.StreamHandler())
proj_logger.setLevel('INFO')

def error_handler(func):

    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as ex:
            proj_logger.error(ex)
            msg = {'message': str(ex)}
            return JSONResponse(content=msg, status_code=500)
    return wrapper

# src.py
from src.project_logger import proj_logger, error_handler

@error_handler
def update_jobs():
    raise Error('Customize Error')

def always_success():
     try:
         raise Error('Nothing')
     except Exception as e:
          proj_logger.error(e)