Метка: fastapi

Долгое время был любителем по изобретать велосипед, и городил собственные реализации API для внешних систем. Да получалось. Да работало. Но если честно, всегда был в этих API бардак. Документации толковой не было, примеров соответственно тоже. Ну правда работал с этими API собственно практически один я, это и спасало от тухлых помидоров 😉

Решил попробовать таки в следующем своем проекте воспользоваться сторонними решениями для документирования и написания API. Выбор пал на FastApi для python. Собственно дальше просто «рыба» с основными фишками которую можно использовать далее расширяя.

Разберем файл main.py:

import uvicorn
from fastapi import FastAPI
from app.routers import hash

tags_metadata = [
    {"name": "use-hash","description": "Создание хэша по логину, паролю и секрету (если есть)"},
    {"name": "delete","description": "Удаление хэга по его  *ID*.",},
]

app = FastAPI(title="OCPI API documentation",description="Инструментарий для работы по протоколу OCPI",
    version="0.0.2",openapi_tags=tags_metadata,)

@app.on_event('startup')
async def startup():
    print("Приложение стартовало")

@app.on_event('shutdown')
async def shutdown():
    print("Приложение закрыто")

app.include_router(hash.router,prefix="/v1")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")

Что происходит в этом скрипте? Сначала инициализируется ядро FastApi:

app = FastAPI(title="OCPI API documentation",description="Инструментарий для работы по протоколу OCPI",
    version="0.0.2",openapi_tags=tags_metadata,)

Т.е. обявляем название API, описание и присваиваем разделы которые будут отображаться в документации http://127.0.0.1:8000/docs

Далее подключаем роутинг из файла /app/routers/hash.py:

app.include_router(hash.router,prefix="/v1")

Этих роутингов естественно может быть не ограниченное количество

Для того чтобы FastApi мог сформировать документацию, необходимо её чуть подготовить. За это отвечают «схемы». Например создадим схему /app/schemas/hash.py:

# -*- coding: utf-8 -*-

from typing import Optional
from pydantic import BaseModel,Field

class hash_out(BaseModel):
    """ Что Выходит """
    hash: str = Field(..., title="Хэш созданый на основе логина пароля", example="iauwhfpseurh e")

class hash_in(BaseModel):
    """ Что входит """
    login: Optional [str] = Field(..., title="Логин пользователя", example="Например Вася")
    password: Optional [str]

class delete_param(BaseModel):
    id : Optional [int]

class delete_answer(BaseModel):
    result : Optional [str]

Т.е. создаем классы с переменными которые могут быть входящими или исходящими пакетами работы API.

Далее создаем непосредственно файл работы с роутом hash:

# -*- coding: utf-8 -*-
"""Different helper-functions to work with users."""
import json

import fastapi
from fastapi import APIRouter,Depends,Request,Header,Response
from app.schemas import hash as models
import hashlib
from typing import Union

router = APIRouter()

@router.post('/hash',response_model=models.hash_out,tags=["use-hash"]) # заданы исходящие параметры
async def hash(into : models.hash_in): # заданы входящие параметры
    # если параметры не заданы
    if into.password==None: into.password="NONE"
    if into.login == None: into.password = "NONE"
    hash=hashlib.sha1((into.login+into.password).encode('utf-8')).hexdigest()
    # сформируем ответ
    out=models.hash_out(hash=hash)
    return out

Что у нас тут? А вот что: мы объявляем, что FastAPI должен перенаправлять весь вход по методу POST с URL /hash (на самом деле /v1/hash, т.к. роутин объявили с префиксом v1) в функцию async def hash. В качестве исходящих значений, в response_model указываем схему hash_out заданную в схемах выше. Так -же установим тэг tags=[«use-hash»], для того чтоб функция API была красиво подписана. В качестве входящих параметров функции hash, указываем соответственно модель с классом hash_in

Если хочется «перехватывать» не только POST запросы, но и GET, то вызов чуть изменится:

@router.get('/hash',response_model=models.hash_out,tags=["use-hash"]) # заданы исходящие параметры
async def hash(into : models.hash_in=Depends()): # заданы входящие параметры.

Т.е. мы изменили вызов на get и в функцию добавили Depends(), т.к. входящие параметры попадают не в виде в данном случае JSON, а в качестве параметров вида: /v1/hash?login=aaa&passowd=bbb, т.е. их нужно конвертировать предварительно в JSON

А что делать, если помимо входящих параметров из POST запроса (ну или GET/PUT/DELETE и т.д), необходимо еще и обработать пришедшие заголовки Headers? Для этого изменим объявление функции, добавив помимо модели, еще и заголовки:

async def hash(into : models.hash_in=Depends(),secret: Union[str, None] = Header(description="Наш секрет",alias="Secret-Code-In",default=None)):

Тогда если в запросе будет присутствовать заголовок Secret-Code-In, то он будет передан в переменную secret

А если нам необходимо не только обработать входящие заголовки, но отдать их при ответе? Тут тогда получиться несколько сложнее.

@router.get('/hash_header',tags=["use-hash"],responses={
        200: {
            "model": models.hash_out,
            "description": "Return has code",
            "headers": {
                "Secret-Code-Out": {"description":"Secret code out","type":"string"}
            }
        }

Т.е. нам придется использовать не обычно используемый response_model, для указания класса со схемой ответа, а response, который позволяет указать шаблон при определенном коде ответа. В результате общий файл роута hash.py получается такой:

# -*- coding: utf-8 -*-
"""Different helper-functions to work with users."""
import json

import fastapi
from fastapi import APIRouter,Depends,Request,Header,Response
from app.schemas import hash as models
import hashlib
from typing import Union

router = APIRouter()

@router.post('/hash',response_model=models.hash_out,tags=["use-hash"]) # заданы исходящие параметры
async def hash(into : models.hash_in): # заданы входящие параметры
    # если параметры не заданы
    if into.password==None: into.password="NONE"
    if into.login == None: into.password = "NONE"
    hash=hashlib.sha1((into.login+into.password).encode('utf-8')).hexdigest()
    # сформируем ответ
    out=models.hash_out(hash=hash)
    return out

@router.get('/hash',response_model=models.hash_out,tags=["use-hash"]) # заданы исходящие параметры
async def hash(into : models.hash_in=Depends()): # заданы входящие параметры.
    # Внимание! Обращаяю внимание на Depends(), оно избавляет от ошибки "Request with GET/HEAD method cannot have body",
    # т.е. так параметры ищутся не в body, а в самом запросе GET в виде параметров

    # если параметры не заданы
    if into.password==None: into.password="NONE"
    if into.login == None: into.password = "NONE"
    hash=hashlib.sha1((into.login+into.password).encode('utf-8')).hexdigest()
    # сформируем ответ
    out=models.hash_out(hash=hash)
    return out

@router.put('/hash',response_model=models.hash_out,tags=["use-hash"]) # заданы исходящие параметры
async def hash(into : models.hash_in): # заданы входящие параметры
    # если параметры не заданы
    if into.password==None: into.password="NONE"
    if into.login == None: into.password = "NONE"
    hash=hashlib.sha1((into.login+into.password).encode('utf-8')).hexdigest()
    # сформируем ответ
    out=models.hash_out(hash=hash)
    return out

@router.delete('/hash',response_model=models.delete_answer,tags=["delete"]) # заданы исходящие параметры
async def hash(into : models.delete_param): # заданы входящие параметры
    # сформируем ответ
    out=models.delete_answer(result="Ok")
    return out

@router.get('/hash_header',tags=["use-hash"],responses={
        200: {
            "model": models.hash_out,
            "description": "Return has code",
            "headers": {
                "Secret-Code-Out": {"description":"Secret code out","type":"string"}
            }
        }
}) # заданы исходящие параметры
async def hash(into : models.hash_in=Depends(),secret: Union[str, None] = Header(description="Наш секрет",alias="Secret-Code-In",default=None)):
    hash = hashlib.sha1((into.login + into.password+secret).encode('utf-8')).hexdigest()
    # сформируем ответ
    out = models.hash_out(hash=hash)
    return Response(status_code=200,content=json.dumps(out.__dict__), media_type="application/json", headers={"Secret-Code-Out": secret})

Что в принципе покрывает используемые в большинстве случаев кейсы использования