FastAPI+Tortoise ORM+FastAPI用戶 Python-關(guān)系-多對(duì)多-魔扣目錄

本文介紹了FastAPI+Tortoise ORM+FastAPI用戶(Python)-關(guān)系-多對(duì)多的處理方法，對(duì)大家解決問(wèn)題具有一定的參考價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)吧！

問(wèn)題描述

我正在使用FastAPI、Tortoise ORM和FastAPI用戶制作一個(gè)API來(lái)學(xué)習(xí)。基本上，我從OpenWeather API獲取城市天氣數(shù)據(jù)，并使用Tortoise ORM存儲(chǔ)在SQLite數(shù)據(jù)庫(kù)中。我對(duì)FastAPI用戶進(jìn)行了身份驗(yàn)證。

我使用的是外鍵，關(guān)系運(yùn)行良好。但現(xiàn)在我想要一些改變(一種改進(jìn))。我希望每個(gè)用戶都有他們的項(xiàng)目，并且在登錄并訪問(wèn)終結(jié)點(diǎn)(@router.get("/me/onlyitems/", response_model=List[schemas.Card_Pydantic]))之后，它只接收他們的項(xiàng)目。

它工作得很好，我只接收用戶項(xiàng)(使用Owner_id)，但我的項(xiàng)表使用外部唯一ID(項(xiàng)是一個(gè)城市，并且每個(gè)城市都有一個(gè)唯一ID)，并且它是主鍵。
但是，當(dāng)另一個(gè)用戶嘗試添加相同的城市時(shí)，它會(huì)給出一個(gè)錯(cuò)誤，因?yàn)楸碇胁荒苡邢嗤某鞘校词故桥c另一個(gè)所有者。嗯，應(yīng)該是這樣的.但我想讓其他用戶訪問(wèn)相同的城市(他們有相同的信息，他們不是定制的)。
然后，我希望多個(gè)用戶能夠訪問(wèn)同一項(xiàng)目(每個(gè)用戶可以訪問(wèn)多個(gè)項(xiàng)目)。

我認(rèn)為為每個(gè)用戶重復(fù)項(xiàng)目信息(使用另一個(gè)ID)并不理想，不同的用戶訪問(wèn)同一項(xiàng)目會(huì)更好。

我認(rèn)為這將是一個(gè)多對(duì)多的關(guān)系，但我不知道如何建立這種關(guān)系并使用FastAPI獲得它。

我想過(guò)創(chuàng)建一個(gè)只有UserID和CityID(以及一個(gè)自動(dòng)遞增ID)的中間表，并使用它來(lái)獲取相關(guān)數(shù)據(jù)，但我不知道這是最好的方法還是可行的。

我嘗試向Tortoise文檔學(xué)習(xí)，但我找不到我需要的部分，也無(wú)法理解文檔中的一些復(fù)雜示例和部分內(nèi)容。

我使用的是包的更新版本和Python3.9.4

好的，讓我展示一些代碼：

Models.py

class UserModel(TortoiseBaseUserModel):  # From FastAPI Users
    cards: fields.ReverseRelation["Card"]


class OAuthAccountModel(TortoiseBaseOAuthAccountModel): # From FastAPI Users
    user = fields.ForeignKeyField("models.UserModel", related_name="oauth_accounts")


class User(models.BaseUser, models.BaseOAuthAccountMixin): # From FastAPI Users
    pass


class UserCreate(models.BaseUserCreate): # From FastAPI Users
    pass


class UserUpdate(User, models.BaseUserUpdate): # From FastAPI Users
    pass


class UserDB(User, models.BaseUserDB, PydanticModel): # From FastAPI Users
    class Config:
        orm_mode = True
        orig_model = UserModel


user_db = TortoiseUserDatabase(UserDB, UserModel, OAuthAccountModel) # From FastAPI Users


class Card(tmodels.Model):
    """
    The Card model, related to the user.
    """
    id = fields.IntField(pk=True)
    city_name = fields.CharField(max_length=30)
    temperature = fields.FloatField()
    state = fields.CharField(max_length=30, null=True)
    icon = fields.CharField(max_length=3, null=True)
    period = fields.CharField(max_length=20, null=True)
    created_at = fields.DatetimeField(auto_now_add=True)
    updated_at = fields.DatetimeField(auto_now=True)
    owner: fields.ForeignKeyRelation[UserModel] = fields.ForeignKeyField(
        "models.UserModel", related_name="cards")

    class PydanticMeta:
        exclude = ["owner"]

我嘗試更改我的所有者字段及其與UserModel的關(guān)系：

class Card(tmodels.Model):
owner = fields.ManyToManyField("models.UserModel", related_name="cards")

class UserModel(TortoiseBaseUserModel):
    cards: fields.ManyToManyRelation["Card"]

我不知道它是否正確，我無(wú)法使其工作，但問(wèn)題可能出在其他地方。

我的用戶終結(jié)點(diǎn)創(chuàng)建項(xiàng)目：

@router.post("/", response_model=Card_Pydantic)
async def create_item_for_current_user(
        card: CardCreate, user: UserDB = Depends(fastapi_users.get_current_active_user)):
    card_obj = await crud.create_user_card(card=card, user=user)
    if card_obj is None:
        raise HTTPException(status_code=404, detail=f"City '{card.city_name}' not found")
    return await card_obj

此處使用的CRUD函數(shù)(用于創(chuàng)建項(xiàng)目)：

async def create_user_card(
        card: schemas.CardCreate, user: UserDB = Depends(fastapi_users.get_current_active_user)):
    card_json = await weather_api.get_api_info_by_name(card.city_name)
    if card_json is None:
        return None
    card_obj = await Card.create(**card_json,
                                 owner_id=user.id)  # ** is used to pass the keys values as kwargs.
    return await card_obj

我的終結(jié)點(diǎn)以獲取用戶擁有的項(xiàng)目：

@router.get("/me/onlyitems/", response_model=List[schemas.Card_Pydantic])
async def read_current_user_items(user: UserDB = Depends(fastapi_users.get_current_active_user)):
    user_cards = await schemas.Card_Pydantic.from_queryset(Card.filter(owner_id=user.id).all())
    return await crud.update_only_card_for(user_cards, user.id)

我的CRUD函數(shù)是：

async def update_only_card_for(user_cards, owner_id):
    for city in user_cards:
        await update_card(city.id, owner_id)
    return await schemas.Card_Pydantic.from_queryset(Card.filter(owner_id=owner_id).all())

# The update function (it will check if it need to update or not)
async def update_card(city_id: int, owner_id: UUID = UserDB):
    card_obj = await Card.get(id=city_id)
    card_time = card_obj.updated_at
    timezone = card_time.tzinfo
    now = datetime.now(timezone)
    time_to_update = now - timedelta(minutes=15)

    if card_time < time_to_update:
        card_json = await weather_api.get_api_info_by_id(city_id)
        await card_obj.update_from_dict(card_json).save()  # with save will update the update_at field
    card_updated = await schemas.Card_Pydantic.from_queryset_single(Card.get(id=city_id))
    return card_updated

如何更改才能獲得用戶需要訪問(wèn)的卡？我認(rèn)為我不能使用按Owner_id篩選(好的，如果我可以創(chuàng)建我的中間表，我可以，但不知道如何做和獲取數(shù)據(jù))，但是我不知道如何建立關(guān)系來(lái)將用戶鏈接到他添加的項(xiàng)目，同時(shí)其他用戶訪問(wèn)相同的項(xiàng)目。

此代碼運(yùn)行良好(我使用的是nuxt.js前端)，直到有重復(fù)的城市…

處理它的最佳方式是什么？如何處理？

謝謝，迭戈。