تطبيقات عالية الأداء مع Python - برنامج تعليمي FastAPI

نشرت: 2022-03-11

تسهل أطر لغة البرمجة الجيدة إنتاج منتجات عالية الجودة بشكل أسرع. حتى أن الأطر العظيمة تجعل تجربة التطوير بأكملها ممتعة. FastAPI هو إطار عمل ويب Python جديد قوي وممتع في الاستخدام. الميزات التالية تجعل FastAPI يستحق المحاولة:

  • السرعة: FastAPI هو أحد أسرع أطر عمل ويب Python. في الواقع ، سرعته متساوية مع Node.js و Go. تحقق من اختبارات الأداء هذه.
  • مستندات مطور مفصلة وسهلة الاستخدام
  • اكتب تلميحًا للرمز الخاص بك واحصل على التحقق من صحة البيانات وتحويلها مجانًا.
  • قم بإنشاء المكونات الإضافية بسهولة باستخدام حقن التبعية.

بناء تطبيق TODO

لاستكشاف الأفكار الكبيرة وراء FastAPI ، دعنا نبني تطبيق TODO ، والذي يُنشئ قوائم مهام لمستخدميه. سيوفر تطبيقنا الصغير الميزات التالية:

  • التسجيل وتسجيل الدخول
  • إضافة عنصر TODO جديد
  • احصل على قائمة بجميع المهام
  • حذف / تحديث عنصر TODO

SQLAlchemy لنماذج البيانات

يحتوي تطبيقنا على نموذجين فقط: User و TODO. بمساعدة SQLAlchemy ، مجموعة أدوات قاعدة البيانات لبايثون ، يمكننا التعبير عن نماذجنا مثل هذا:

 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 الخاصة بنا. لكتابة نماذج pydantic ، فكر في جميع الطرق التي ستتدفق بها معلومات المستخدم و 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

هنا ، قمنا بتمييز اسم العائلة والاسم الأول وكلمة المرور كسلسلة ، ولكن يمكن تشديدها بشكل أكبر باستخدام سلاسل مقيدة مقيدة تتيح عمليات التحقق مثل الحد الأدنى للطول والحد الأقصى للطول والتعبيرات الرسمية.

لدعم إنشاء وإدراج عناصر 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 ، نحتاج إلى إنشاء JWTs وكذلك فك تشفيرها للحصول على بيانات اعتماد المستخدم. نحدد الوظائف التالية لتوفير هذه الوظيفة.

 # 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"}

استخدام حقن التبعية للوصول إلى قاعدة البيانات وحماية نقاط النهاية

لقد قمنا بإعداد نقطة نهاية تسجيل الدخول التي توفر JWT للمستخدم عند تسجيل الدخول بنجاح. يمكن للمستخدم حفظ هذا الرمز المميز في التخزين المحلي وإظهاره إلى نهايتنا الخلفية كرأس ترخيص. يمكن لنقاط النهاية التي تتوقع الوصول فقط من المستخدمين الذين قاموا بتسجيل الدخول أن تقوم بفك تشفير الرمز المميز ومعرفة من هو الطالب. لا يرتبط هذا النوع من العمل بنقطة نهاية معينة ، بل هو منطق مشترك يستخدم في جميع نقاط النهاية المحمية. من الأفضل إعداد منطق فك التشفير باعتباره تبعية يمكن استخدامها في أي معالج طلب.

في FastAPI-talk ، تعتمد وظائف عملية المسار (معالجات الطلب) على get_current_user . تحتاج تبعية get_current_user إلى اتصال بقاعدة البيانات وربطها بمنطق 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 Create، Read، Update، Delete (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 التي تستند إلى مكتبة الطلبات الشهيرة ، ويمكننا إجراء الاختبارات باستخدام 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 حديثة. استمتع.