Технология

Високоефективни приложения с Python - урок за FastAPI

Добрите рамки на езика за програмиране улесняват по-бързото производство на качествени продукти. Страхотните рамки дори правят цялото преживяване за разработка приятно. FastAPI е нова Python уеб рамка, която е мощна и приятна за използване. Следните функции правят FastAPI заслужаващо да се опита:

Скорост: FastAPI е една от най-бързите Python уеб рамки. Всъщност скоростта му е равна на Node.js и Go. Проверете тези тестове за ефективност.
Подробни и лесни за използване документи за разработчици
Въведете намек за кода си и получете безплатна проверка и преобразуване на данни.
Създавайте приставки лесно с помощта на инжектиране на зависимости.

Изграждане на приложение TODO

За да изследваме големите идеи зад FastAPI, нека създадем приложение TODO, което създава списъци със задачи за своите потребители. Нашето малко приложение ще предоставя следните функции:

Регистрация и влизане
Добавете нов елемент TODO
Получете списък на всички TODO
Изтрийте / актуализирайте TODO елемент

SQLAlchemy за модели на данни

Нашето приложение има само два модела: User и TODO. С помощта на SQLAlchemy, набора от инструменти за база данни за Python, можем да изразим нашите модели по следния начин:

class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True, index=True) lname = Column(String) fname = Column(String) email = Column(String, unique=True, index=True) todos = relationship('TODO', back_populates='owner', cascade='all, delete-orphan') class TODO(Base): __tablename__ = 'todos' id = Column(Integer, primary_key=True, index=True) text = Column(String, index=True) completed = Column(Boolean, default=False) owner_id = Column(Integer, ForeignKey('users.id')) owner = relationship('User', back_populates='todos')

След като нашите модели са готови, нека напишем конфигурационния файл за SQLAlchemy, за да знае как да установи връзка с базата данни.

import os from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL = os.environ['SQLALCHEMY_DATABASE_URL'] engine = create_engine( SQLALCHEMY_DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()

Разгърнете силата на подсказките за типа

Значителна част от всеки API проект се отнася до рутинните неща като валидиране и преобразуване на данни. Нека да се справим с нея отпред, преди да преминем към писането на обработващи заявки. С FastAPI ние изразяваме схемата на нашите входящи / изходящи данни, използвайки pydantic модели и след това използваме тези pydantic модели, за да въведем намек и да се насладим на безплатна проверка и преобразуване на данни. Моля, обърнете внимание, че тези модели не са свързани с работния процес на нашата база данни и посочват само формата на данните, които се вливат и излизат от нашия REST интерфейс. За да пишете пидантски модели, помислете за всички начини, по които информацията за потребителя и TODO ще се влива и изтича.

Традиционно нов потребител ще се регистрира за нашата услуга TODO и съществуващ потребител ще влезе в системата. И двете взаимодействия се занимават с потребителска информация, но формата на данните ще бъде различна. Нуждаем се от повече информация от потребителите по време на регистрация и минимална (само имейл и парола) при влизане. Това означава, че се нуждаем от два пидантски модела, за да изразим тези две различни форми на потребителска информация.

В нашето приложение TODO обаче ще използваме вградената поддръжка на OAuth2 в FastAPI за базиран на JSON Web Tokens (JWT) поток за вход. Просто трябва да дефинираме UserCreate схема тук, за да укажете данни, които ще се вливат в нашата крайна точка за регистрация и UserBase схема, която да се върне като отговор в случай, че процесът на регистрация е успешен.

from pydantic import BaseModel from pydantic import EmailStr class UserBase(BaseModel): email: EmailStr class UserCreate(UserBase): lname: str fname: str password: str

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

За да подпомогнем създаването и изброяването на TODO елементи, ние дефинираме следната схема:

class TODOCreate(BaseModel): text: str completed: bool

За да поддържаме актуализацията на съществуващ TODO елемент, ние дефинираме друга схема:

class TODOUpdate(TODOCreate): id: int

С това приключихме с дефинирането на схеми за всички обмени на данни. Сега насочваме вниманието си към обработчици на заявки, където тези схеми ще бъдат използвани за извършване на всички тежки операции за преобразуване и проверка на данни безплатно.

Първо, нека позволим на потребителите да се регистрират, тъй като всички наши услуги трябва да имат достъп от удостоверен потребител. Пишем първия си манипулатор на заявки, използвайки UserCreate и UserBase схема, дефинирана по-горе.

@app.post('/api/users', response_model=schemas.User) def signup(user_data: schemas.UserCreate, db: Session = Depends(get_db)): '''add new user''' user = crud.get_user_by_email(db, user_data.email) if user: raise HTTPException(status_code=409, detail='Email already registered.') signedup_user = crud.create_user(db, user_data) return signedup_user

В тази кратка част от кода се случва много. Използвали сме декоратор, за да посочим HTTP глагола, URI и схемата на успешните отговори. За да се уверим, че потребителят е изпратил правилните данни, ние въведохме намек за тялото на заявката с по-рано дефиниран UserCreate схема. Методът дефинира друг параметър за получаване на манипулатор в базата данни - това е инжектиране на зависимост в действие и е обсъдено по-късно в този урок.

Защита на нашия API

Искаме следните функции за сигурност в нашето приложение:

Хеширане на парола
JWT-базирано удостоверяване

За хеширане на пароли можем да използваме Passlib. Нека дефинираме функции, които се справят с хеширането на паролата и проверяват дали паролата е правилна.

from passlib.context import CryptContext pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto') def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password) def authenticate_user(db, email: str, password: str): user = crud.get_user_by_email(db, email) if not user: return False if not verify_password(password, user.hashed_password): return False return user

За да активираме JWT-базирано удостоверяване, трябва да генерираме JWT, както и да ги декодираме, за да получим потребителски идентификационни данни. Ние дефинираме следните функции, за да предоставим тази функционалност.

# install PyJWT import jwt from fastapi.security import OAuth2PasswordBearer SECRET_KEY = os.environ['SECRET_KEY'] ALGORITHM = os.environ['ALGORITHM'] def create_access_token(*, data: dict, expires_delta: timedelta = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({'exp': expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def decode_access_token(db, token): credentials_exception = HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail='Could not validate credentials', headers={'WWW-Authenticate': 'Bearer'}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) email: str = payload.get('sub') if email is None: raise credentials_exception token_data = schemas.TokenData(email=email) except PyJWTError: raise credentials_exception user = crud.get_user_by_email(db, email=token_data.email) if user is None: raise credentials_exception return user

Сега ще дефинираме крайна точка за вход и ще приложим потока на паролата OAuth2. Тази крайна точка ще получи имейл и парола. Ще проверим идентификационните данни спрямо базата данни и при успех ще издадем JSON уеб токен на потребителя.

За да получим идентификационните данни, ще използваме OAuth2PasswordRequestForm, което е част от помощните програми за защита на FastAPI.

@app.post('/api/token', response_model=schemas.Token) def login_for_access_token(db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()): '''generate access token for valid credentials''' user = authenticate_user(db, form_data.username, form_data.password) if not user: raise HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail='Incorrect email or password', headers={'WWW-Authenticate': 'Bearer'}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token(data={'sub': user.email}, expires_delta=access_token_expires) return {'access_token': access_token, 'token_type': 'bearer'}

Използване на инжекция на зависимост за достъп до DB и защита на крайни точки

Настроихме крайната точка за вход, която предоставя JWT на потребител при успешно влизане. Потребителят може да запази този токен в локално хранилище и да го покаже на задния ни край като заглавка за оторизация. Крайните точки, които очакват достъп само от влезли потребители, могат да декодират маркера и да разберат кой е заявителят. Този вид работа не е обвързана с определена крайна точка, а по-скоро е споделена логика, използвана във всички защитени крайни точки. Най-добре е да настроите логиката за декодиране на маркера като зависимост, която може да се използва във всеки манипулатор на заявки.

В FastAPI-говорите, нашите функции за работа на пътя (обработващи заявки) тогава ще зависят от get_current_user. get_current_user зависимостта трябва да има връзка с базата данни и да се свърже с FastAPI’s OAuth2PasswordBearer логика за получаване на жетон. Ще разрешим този проблем, като направим get_current_user зависят от други функции. По този начин можем да определим вериги на зависимост, което е много мощна концепция.

def get_db(): '''provide db session to path operation functions''' try: db = SessionLocal() yield db finally: db.close() def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)): return decode_access_token(db, token) @app.get('/api/me', response_model=schemas.User) def read_logged_in_user(current_user: models.User = Depends(get_current_user)): '''return user settings for current user''' return current_user

Влезлите потребители могат да CRUD TODOs

Преди да напишем функциите за операция на пътя за TODO Създаване, четене, актуализиране, изтриване (CRUD), ние дефинираме следните помощни функции за изпълнение на действителни CRUD на db.

def create_todo(db: Session, current_user: models.User, todo_data: schemas.TODOCreate): todo = models.TODO(text=todo_data.text, completed=todo_data.completed) todo.owner = current_user db.add(todo) db.commit() db.refresh(todo) return todo def update_todo(db: Session, todo_data: schemas.TODOUpdate): todo = db.query(models.TODO).filter(models.TODO.id == id).first() todo.text = todo_data.text todo.completed = todo.completed db.commit() db.refresh(todo) return todo def delete_todo(db: Session, id: int): todo = db.query(models.TODO).filter(models.TODO.id == id).first() db.delete(todo) db.commit() def get_user_todos(db: Session, userid: int): return db.query(models.TODO).filter(models.TODO.owner_id == userid).all()

Тези функции на ниво db ще се използват в следните REST крайни точки:

@app.get('/api/mytodos', response_model=List[schemas.TODO]) def get_own_todos(current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): '''return a list of TODOs owned by current user''' todos = crud.get_user_todos(db, current_user.id) return todos @app.post('/api/todos', response_model=schemas.TODO) def add_a_todo(todo_data: schemas.TODOCreate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): '''add a TODO''' todo = crud.create_meal(db, current_user, meal_data) return todo @app.put('/api/todos/{todo_id}', response_model=schemas.TODO) def update_a_todo(todo_id: int, todo_data: schemas.TODOUpdate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): '''update and return TODO for given id''' todo = crud.get_todo(db, todo_id) updated_todo = crud.update_todo(db, todo_id, todo_data) return updated_todo @app.delete('/api/todos/{todo_id}') def delete_a_meal(todo_id: int, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): '''delete TODO of given id''' crud.delete_meal(db, todo_id) return {'detail': 'TODO Deleted'}

Напишете тестове

Нека напишем няколко теста за нашия TODO API. FastAPI предоставя TestClient клас, който се основава на популярната библиотека Requests и можем да стартираме тестовете с Pytest.

За да сме сигурни, че само влезли потребители могат да създадат TODO, можем да напишем нещо подобно:

from starlette.testclient import TestClient from .main import app client = TestClient(app) def test_unauthenticated_user_cant_create_todos(): todo=dict(text='run a mile', completed=False) response = client.post('/api/todos', data=todo) assert response.status_code == 401

Следващият тест проверява нашата крайна точка за вход и генерира JWT, ако е представен с валидни идентификационни данни за вход.

def test_user_can_obtain_auth_token(): response = client.post('/api/token', data=good_credentials) assert response.status_code == 200 assert 'access_token' in response.json() assert 'token_type' in response.json()

Обобщавайки го

Завършихме внедряването на много просто приложение TODO с помощта на FastAPI. Досега видяхте силата на подсказките за типа, използвани добре за определяне на формата на входящите и изходящите данни чрез нашия REST интерфейс. Дефинираме схемите на едно място и го оставяме на FastAPI да приложи валидиране и преобразуване на данни. Другата забележителна характеристика е инжектирането на зависимост. Използвахме тази концепция за пакетиране на споделената логика за получаване на връзка с база данни, декодиране на JWT, за да се получи влезлият в момента потребител и внедряване на прост OAuth2 с парола и носител. Видяхме също как зависимостите могат да бъдат оковани заедно.

Можем лесно да приложим тази концепция, за да добавим функции като ролеви достъп. Освен това пишем кратък и мощен код, без да научаваме особеностите на рамката. С прости думи, FastAPI е колекция от мощни инструменти, които не е нужно да научавате, защото те са просто модерен Python. Забавлявай се.

Разбиране на основите

Какво е FastAPI?

FastAPI е рамка на Python и набор от инструменти, които позволяват на разработчиците да използват REST интерфейс за извикване на често използвани функции за внедряване на приложения. Той е достъпен чрез REST API за извикване на общи градивни елементи за приложение. В този пример авторът използва FastAPI за създаване на акаунти, влизане и удостоверяване.

Как стартирате FastAPI?

Както всички REST интерфейси, FastAPI се извиква от вашия код. Той предлага функции като подсказване на типа за предадени данни, инжектиране на зависимост и удостоверяване, така че да не се налага да пишете свои собствени функции.

Готов ли е за производство FastAPI?

Докато рамка с отворен код, FastAPI е напълно готова за производство, с отлична документация, поддръжка и лесен за използване интерфейс. Може да се използва за изграждане и стартиране на приложения, които са толкова бързи, колкото тези, написани на други скриптови езици.