Skip to content

exhibits#

Exhibits Router app

create_exhibit(exhibit, session=Depends(get_async_session)) async #

Create a new exhibit in the database

Source code in zoo/api/exhibits.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
@exhibits_router.post("/exhibits", response_model=ExhibitsRead)
async def create_exhibit(
    exhibit: ExhibitsCreate, session: AsyncSession = Depends(get_async_session)
) -> ExhibitsRead:
    """
    Create a new exhibit in the database
    """
    new_exhibit = Exhibits(**exhibit.model_dump(exclude_unset=True))
    session.add(new_exhibit)
    await session.commit()
    await session.refresh(new_exhibit)
    exhibit_model = ExhibitsRead.model_validate(new_exhibit)
    return exhibit_model

delete_exhibit(exhibit_id, session=Depends(get_async_session)) async #

Delete exhibit from the database

Source code in zoo/api/exhibits.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@exhibits_router.delete("/exhibits/{exhibit_id}", response_model=ExhibitsRead)
async def delete_exhibit(
    exhibit_id: int,
    session: AsyncSession = Depends(get_async_session),
) -> ExhibitsRead:
    """
    Delete exhibit from the database
    """
    exhibit: Optional[Exhibits] = await session.get(Exhibits, exhibit_id)
    exhibit = check_model(model_instance=exhibit, model_class=Exhibits, id=exhibit_id)
    exhibit.deleted_at = func.current_timestamp()
    session.add(exhibit)
    await session.commit()
    await session.refresh(exhibit)
    exhibit_model = ExhibitsRead.model_validate(exhibit)
    return exhibit_model

get_exhibit(exhibit_id, session=Depends(get_async_session)) async #

Get exhibit from the database

Source code in zoo/api/exhibits.py 
63
64
65
66
67
68
69
70
71
72
73
74
@exhibits_router.get("/exhibits/{exhibit_id}", response_model=ExhibitsRead)
async def get_exhibit(
    exhibit_id: int,
    session: AsyncSession = Depends(get_async_session),
) -> ExhibitsRead:
    """
    Get exhibit from the database
    """
    exhibit: Optional[Exhibits] = await session.get(Exhibits, exhibit_id)
    exhibit = check_model(model_instance=exhibit, model_class=Exhibits, id=exhibit_id)
    exhibit_model = ExhibitsRead.model_validate(exhibit)
    return exhibit_model

get_exhibit_animals(exhibit_id, session=Depends(get_async_session)) async #

List animals in an exhibit

Source code in zoo/api/exhibits.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@exhibits_router.get("/exhibits/{exhibit_id}/animals", response_model=List[AnimalsRead])
async def get_exhibit_animals(
    exhibit_id: int,
    session: AsyncSession = Depends(get_async_session),
) -> List[AnimalsRead]:
    """
    List animals in an exhibit
    """
    exhibit: Optional[Exhibits] = await session.get(
        entity=Exhibits,
        ident=exhibit_id,
        options=[
            joinedload(
                Exhibits.animals
            )  # explicit load of relationship supports async session
        ],
    )
    exhibit = check_model(model_instance=exhibit, model_class=Exhibits, id=exhibit_id)
    animals: List[Animals] = exhibit.animals
    animals_models = [AnimalsRead.model_validate(animal) for animal in animals]
    return animals_models

get_exhibit_staff(exhibit_id, session=Depends(get_async_session)) async #

List staff in an exhibit

Source code in zoo/api/exhibits.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
@exhibits_router.get("/exhibits/{exhibit_id}/staff", response_model=List[StaffRead])
async def get_exhibit_staff(
    exhibit_id: int,
    session: AsyncSession = Depends(get_async_session),
) -> List[StaffRead]:
    """
    List staff in an exhibit
    """
    exhibit: Optional[Exhibits] = await session.get(
        entity=Exhibits,
        ident=exhibit_id,
        options=[
            joinedload(Exhibits.staff),
        ],
    )
    exhibit = check_model(model_instance=exhibit, model_class=Exhibits, id=exhibit_id)
    staff: List[Staff] = exhibit.staff
    staff_models = [StaffRead.model_validate(staff) for staff in staff]
    return staff_models

get_exhibits(offset=0, limit=Query(default=100, le=100), session=Depends(get_async_session)) async #

Get exhibits from the database

Source code in zoo/api/exhibits.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@exhibits_router.get("/exhibits", response_model=List[ExhibitsRead])
async def get_exhibits(
    offset: int = 0,
    limit: int = Query(default=100, le=100),
    session: AsyncSession = Depends(get_async_session),
) -> List[ExhibitsRead]:
    """
    Get exhibits from the database
    """
    result = await session.execute(
        select(Exhibits)
        .where(Exhibits.deleted_at.is_(None))
        .order_by(Exhibits.id)
        .offset(offset)
        .limit(limit)
    )
    exhibits: Sequence[Exhibits] = result.scalars().all()
    exhibit_models = [ExhibitsRead.model_validate(exhibit) for exhibit in exhibits]
    return exhibit_models

update_exhibit(exhibit_id, exhibit, session=Depends(get_async_session)) async #

Update exhibit from the database

Source code in zoo/api/exhibits.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@exhibits_router.patch("/exhibits/{exhibit_id}", response_model=ExhibitsRead)
async def update_exhibit(
    exhibit_id: int,
    exhibit: ExhibitsUpdate,
    session: AsyncSession = Depends(get_async_session),
) -> ExhibitsRead:
    """
    Update exhibit from the database
    """
    db_exhibit: Optional[Exhibits] = await session.get(Exhibits, exhibit_id)
    db_exhibit = check_model(
        model_instance=db_exhibit, model_class=Exhibits, id=exhibit_id
    )
    for field, value in exhibit.model_dump(exclude_unset=True).items():
        setattr(db_exhibit, field, value)
    session.add(db_exhibit)
    await session.commit()
    await session.refresh(db_exhibit)
    exhibit_model = ExhibitsRead.model_validate(db_exhibit)
    return exhibit_model