Chapter 1

配置

配合vscode,配置 pylint ,防止出现各种问题

.vscode/setting.json

{
    "pylint.cwd": "${workspaceFolder}/python-server",
    "python.defaultInterpreterPath": "${workspaceFolder}/python-server/.venv/bin/python3"
}

设置 py项目的根目录为项目子目录

在根目录中的 .env 文件中添加:

PYTHONPATH=<subdir>

第三方模块

fastapi

mongo

安装

poetry add pymongo

使用

初始化

import pymongo # 连接到 MongoDB 数据库 
client = pymongo.MongoClient("mongodb://root:root@192.168.0.102:27017/") # 替换为你的 MongoDB 服务器地址 # 选择数据库 
db = client["mydatabase"] # 替换为你的数据库名称 # 选择集合 
collection = db["mycollection"] # 替换为你的集合名称

fastapi

快速开始

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()
origins = [
    "*"
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/")
async def root():
    return {"message": "Hello World"}

控制台中执行命令

# fastapi dev src.main.py # dev
uvicorn src.main:app --reload # prod 去掉 --reload

各种参数

Query

@app.get("/items/")
async def read_items(q: str | None = None):

ORM

对象相关

model 和 schema 转换

class Item(BaseModel):

    # 不确定是哪个属性
    class   Config:
        orm_mode = True  
        from_attributes=True

日志

日志库

loguru

loki

# example of usage grafana/loki api when you need push any log/message from your python scipt
import requests
import json
import datetime
from src.config.config import logger


def push(msg, host=None):
    curr_datetime = datetime.datetime.now()
    curr_datetime.timestamp()*10**9

    # push msg log into grafana-loki
    url = 'http://192.168.0.102:3100/loki/api/v1/push'
    headers = {
        'Content-type': 'application/json'
    }

    curr_nanoseconds = f"{int(curr_datetime.timestamp()*10**9)}"
    payload = {
        "streams": [
            {
                "stream": {
                    "server": "dns-api"
                },
                "values": [
                    [f"{curr_nanoseconds}", msg]
                ]
            }
        ]
    }

    payload_str = json.dumps(payload)
    response = requests.post(url, data=payload_str, headers=headers)
    logger.debug(
        f"Request: {payload_str} \nResonse : {response.ok} {response.text}")
    return response.ok

自定义模块

附录

1. Playground