Build a Cocktail API with Beanie and MongoDB

1
{
2
   "_id": "5f7daa158ec9dfb536781b0a",
3
   "name": "Hunter's Moon",
4
   "ingredients": [
5
      {
6
         "name": "Vermouth",
7
         "quantity": {
8
            "quantity": "25",
9
            "unit": "ml"
10
         }
11
      },
12
      {
13
         "name": "Maraschino Cherry",
14
         "quantity": {
15
            "quantity": "15",
16
            "unit": "ml"
17
         }
18
      },
19
      {
20
         "name": "Sugar Syrup",
21
         "quantity": {
22
            "quantity": "10",
23
            "unit": "ml"
24
         }
25
      },
26
      {
27
         "name": "Lemonade",
28
         "quantity": {
29
            "quantity": "100",
30
            "unit": "ml"
31
         }
32
      },
33
      {
34
         "name": "Blackberries",
35
         "quantity": {
36
            "quantity": "2",
37
            "unit": null
38
         }
39
      }
40
   ]
41
}

1
class Cocktail(Document):
2
    class Settings:
3
        name = "recipes"
4
5
    name: str
6
    ingredients: List["Ingredient"]
7
    instructions: List[str]
8
9
10
class Ingredient(BaseModel):
11
    name: str
12
    quantity: Optional["IngredientQuantity"]
13
14
15
class IngredientQuantity(BaseModel):
16
    quantity: Optional[str]
17
    unit: Optional[str]
18
19
20
class IngredientAggregation(BaseModel):
21
    """ A model for an ingredient count. """
22
23
    id: str = Field(None, alias="_id")
24
    total: int

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
# ... imports skipped
2
3
cocktail_router = APIRouter()

1
@cocktail_router.get("/cocktails/", response_model=List[Cocktail])
2
async def list_cocktails():
3
   return await Cocktail.find_all().to_list()

1
just run

1
uvicorn beaniecocktails:app --reload

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
[
2
   {"_id":"7-Up","total":1},
3
   {"_id":"Amaretto","total":2},
4
   {"_id":"Angostura Bitters","total":1},
5
   {"_id":"Apple schnapps","total":1},
6
   {"_id":"Applejack","total":1},
7
   {"_id":"Apricot brandy","total":1},
8
   {"_id":"Bailey","total":1},
9
   {"_id":"Baileys irish cream","total":1},
10
   {"_id":"Bitters","total":3},
11
   {"_id":"Blackberries","total":1},
12
   {"_id":"Blended whiskey","total":1},
13
   {"_id":"Bourbon","total":1},
14
   {"_id":"Bourbon Whiskey","total":1},
15
   {"_id":"Brandy","total":7},
16
   {"_id":"Butterscotch schnapps","total":1},
17
]

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
   ]