FastAPI - SqlMODEL

Docs: 

• https://fastapi.tiangolo.com/tutorial/sql-databases 

• https://sqlmodel.tiangolo.com 

SQL Model is SQL Database for FastAPI

Cài đặt thư viện: 

pip install sqlmodel

pip install pymysql

Chuẩn bị CSDL MySQL, tạo sẵn database ví dụ 2421comp103101

Khai báo chuỗi kết nối, ở đây dùng PyMySQL. Bạn có thể xem các driver cho các database khác ở trang chính SQLAlchemy SQLAlchemy là một ORM (Object-Relational Mapping) cho ngôn ngữ lập trình Python.

ORM là gì?

ORM (Object Relation Mapping) là một kỹ thuật trong lập trình giúp biểu diễn các dòng dữ liệu (record) trong cơ sở dữ liệu bằng các đối tượng, vật thể (object) tương ứng trong ngôn ngữ lập trình. Nhờ đó, ta có thể tương tác, xử lí những data record tương tự như những object. 

Chuỗi kết nối.

engine = create_engine("mysql+pymysql://username:password@hostname/database_name")

Ví dụ:

engine = create_engine("mysql+pymysql://root:@localhost/2421comp103101")

ở trên dùng MySQL của gói XAMPP nên username=root, password rỗng.

Nếu bạn dùng CSDL khác có thể tham khảo ở https://docs.sqlalchemy.org/en/20/core/engines.html 

Định nghĩa Model User

Khi ứng dụng chạy sẽ map với table User.

class User(SQLModel, table=True):

    id: int | None = Field(default=None, primary_key=True)

    username: str = Field(index=True)

    password: str

    role: int = Field(default=None)

    is_active: bool | None = Field(default=True)

Định nghĩa một số hàm liên quan như tạo table nếu chưa có, đánh dấu session:

def create_db_and_tables():

    SQLModel.metadata.create_all(engine)

def get_session():

    with Session(engine) as session:

        yield session

SessionDep = Annotated[Session, Depends(get_session)]

Khai báo sự kiện start_up:

@app.on_event("startup")

def on_startup():

    create_db_and_tables()

Thư viện cần khai báo:

from fastapi import FastAPI, UploadFile, File, Depends, Query 

from sqlmodel import Field, Session, SQLModel, create_engine, select

from typing import Annotated

Tạo các API CRUD cho user

@app.post("/users/", tags=["user"])

def create_user(model: User, session: SessionDep) -> User:

    session.add(model)

    session.commit()

    session.refresh(model)

    return model

@app.get("/users/", tags=["user"])

def read_users(

    session: SessionDep,

    offset: int = 0,

    limit: Annotated[int, Query(le=100)] = 100,

) -> list[User]:

    users = session.exec(

        select(User).offset(offset).limit(limit)

    ).all()

    return users

@app.get("/users/{user_id}", tags=["user"])

def read_user(user_id: int, session: SessionDep) -> User:

    user = session.get(User, user_id)

    if not user:

        raise HTTPException(

            status_code=404,

            detail=f"User {user_id} not found"

        )

    return user

@app.delete("/users/{user_id}", tags=["user"])

def delete_user(user_id: int, session: SessionDep):

    user = session.get(User, user_id)

    if not user:

        raise HTTPException(

            status_code=404,

            detail=f"User {user_id} not found"

        )

    try:

        session.delete(user)

        session.commit()

        return {"success": True}

    except Exception as ex:

        print(ex)

        return {"success": False}

• SV tự viết cho API PUT.

• Phân biệt API PUT vs PATCH

• Thử tạo model cho các bảng Loai, HangHoa, trong đó một loại có nhiều hàng hóa.

class Loai(SQLModel, table=True):    

    MaLoai: int | None = Field(default=None, primary_key=True)

    TenLoai: str = Field(index=True)

class HangHoa(SQLModel, table=True):    

    MaHH: int | None = Field(default=None, primary_key=True)

    TenHH: str = Field(index=True)

    Hinh: str = Field()

    MaLoai: int | None =  Field(foreign_key="loai.MaLoai")

Tham khảo thêm phần định nghĩa quan hệ giữa các bảng: https://sqlmodel.tiangolo.com/tutorial/relationship-attributes/ 

• Thực hiện filter data (tham khảo https://sqlmodel.tiangolo.com/tutorial/where/)