Crie uma Cocktail API com o Beanie e o MongoDB

1
# ... some code skipped
2
3
async def app_lifespan(app: FastAPI):
4
    # startup code goes here:
5
    client: AsyncIOMotorClient = AsyncIOMotorClient(
6
        Settings().mongodb_url,
7
    )
8
    await init_beanie(client.get_default_database(), document_models=[Cocktail])
9
    app.include_router(cocktail_router, prefix="/v1")
10
11
    yield
12
13
    # shutdown code goes here:
14
    client.close()
15
16
app = FastAPI(lifespan=app_lifespan)
17
18
# I'm using pydantic_settings to load the MongoDB connection string from an environment variable,
19
# or it defaults to a local installation.
20
class Settings(BaseSettings):
21
    mongodb_url: str = "mongodb://localhost:27017/cocktails"

1
async def get_cocktail(cocktail_id: PydanticObjectId) -> Cocktail:
2
   """ Helper function to look up a cocktail by id """
3
4
   cocktail = await Cocktail.get(cocktail_id)
5
   if cocktail is None:
6
      raise HTTPException(status_code=404, detail="Cocktail not found")
7
   return cocktail
8
9
@cocktail_router.get("/cocktails/{cocktail_id}", response_model=Cocktail)
10
async def get_cocktail_by_id(cocktail: Cocktail = Depends(get_cocktail)):
11
   return cocktail

1
# models.py:
2
3
class IngredientAggregation(BaseModel):
4
   """ A model for an ingredient count. """
5
6
   id: str = Field(None, alias="_id")
7
   total: int
8
9
# routes.py:
10
11
@cocktail_router.get("/ingredients", response_model=List[IngredientAggregation])
12
async def list_ingredients():
13
   """ Group on each ingredient name and return a list of `IngredientAggregation`s. """
14
15
   return await Cocktail.aggregate(
16
      aggregation_query=[
17
         {"$unwind": "$ingredients"},
18
         {"$group": {"_id": "$ingredients.name", "total": {"$sum": 1}}},
19
         {"$sort": {"_id": 1}},
20
      ],
21
      projection_model=IngredientAggregation,
22
   ).to_list()

1
@cocktail_router.get("/cocktail_autocomplete", response_model=List[str])
2
async def cocktail_autocomplete(fragment: str):
3
   """ Return an array of cocktail names matched from a string fragment. """
4
5
   return [
6
      c["name"]
7
      for c in await Cocktail.aggregate(
8
         aggregation_query=[
9
            {
10
               "$search": {
11
                  "autocomplete": {
12
                     "query": fragment,
13
                     "path": "name",
14
                  }
15
               }
16
            }
17
         ]
18
      ).to_list()
19
   ]