Учебное пособие по удалению событий API с помощью SQL ORM - UPROGER

В настоящее время сертификаты встречаются повсеместно. Документирование небольших проектов по повышению квалификации, которыми можно впоследствии похвастаться, – лучший способ получить признание в качестве профессионального инженера.

Задачи TL;DR

Загрузите предоставленный код для решения задачи.

Задача 1: Реализация функции отмены билетов

Добавьте новый API-маршрут или конечную точку для обработки отмены билета на основе его ID.
Перед тем как разрешить отмену, проверьте, не началось ли событие, связанное с билетом.
Убедитесь, что билет удаляется из базы данных, если отмена прошла успешно.

Задача 2: Реализовать функцию изменения имени билета

Добавьте еще один маршрут или конечную точку API для изменения имени билета на основе его ID.
Перед тем как разрешить изменение имени, проверьте, не началось ли еще событие для билета.
Обновление имени билета в базе данных, если изменение имени действительно.

Задача 3: Обработка связанных тикетов при удалении события

Модифицируйте API-маршрут или конечную точку для удаления события, чтобы также удалять все связанные с ним билеты.
Убедитесь, что при удалении события и билетов на него они удаляются из базы данных.

Задача 4: Модульные тесты для функции аннулирования билетов

Напишите модульные тесты для проверки корректности работы отмены билетов.
Включите тесты для проверки того, что в API выполняется условие запрета отмены билетов после начала мероприятия.

Задача 5: Модульные тесты для функции изменения имени билета

Создайте модульные тесты для проверки функциональности изменения имени в билете.
Убедитесь, что API правильно проверяет и предотвращает изменение имени после начала события.

Задача 6: Модульные тесты для связанных тикетов при удалении события

Написать модульные тесты, подтверждающие удаление всех связанных с событием тикетов при его удалении.
Протестируйте сценарии с несколькими билетами, связанными с событием, чтобы убедиться, что все они удалены.

Задача 7: Протестировать ограничение времени начала события

Создайте тесты для проверки того, ограничена ли отмена билетов и изменение имени после времени начала мероприятия.
Протестируйте оба случая на предмет действительных и недействительных изменений/отмен на основе времени начала мероприятия.

Задача 8: Обеспечить обработку ошибок и ответных сообщений

Реализовать соответствующую обработку ошибок при недействительных запросах (например, при попытке отменить/обменять билет после начала мероприятия).
Обеспечьте четкие и информативные ответные сообщения для различных сценариев.

Задача 9: Протестировать весь API с новой функциональностью

Запуск и проверка всего API с добавлением функций отмены билетов и изменения имени.
Тестирование различных сценариев и граничных ситуаций для обеспечения корректности и надежности API.

Решение

Мы просто добавляем новые возможности в наш API – в этой задаче давайте добавим функцию отмены и обновления. Вот как я реализовал эти изменения, сначала на операционном уровне:

# after
def delete_ticket(ticket_id: int, database: Database) -> Ticket:
    ticket = get_ticket(ticket_id, database)
    event = get_event(ticket.event_id, database)
    
    if event.has_started():
        raise EventAlreadyStarted(
            f"Event with id {ticket.event_id} has already started."
        )

    database.delete(ticket)
    event.available_tickets += 1
    
    database.commit()
    return ticket

def update_ticket(ticket_id: int, ticket_update: TicketUpdate, database: Database) -> Ticket:
    ticket = get_ticket(ticket_id, database)
    event = get_event(ticket.event_id, database)
    
    if event.has_started():
        raise EventAlreadyStarted(
            f"Event with id {ticket.event_id} has already started."
        )

    if ticket_update.customer_name:
        ticket.customer_name = ticket_update.customer_name
    database.commit()

Для обновления билетов мы можем создать новый объект базы данных TicketUpdate и отделить его от операций. И get_ticket, и get_event имеют свои собственные ошибки NotFoundError, которые мы хорошо продемонстрировали в последней задаче (посмотрите, если хотите перевести обработку ошибок без ответственности за сопряжение). Для дальнейшего поддержания низкого уровня сопряжения я добавил ошибку EventAlreadyStarted, которая возникает, когда четное событие уже началось. Как это реализовать?

Самое замечательное в SQLalchemy ORM то, что мы можем встраивать методы непосредственно в модель!

# Define event model
class Event(Base):
    __tablename__ = "events"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String, index=True)
    location: Mapped[str] = mapped_column(String, index=True)
    start_date: Mapped[datetime] = mapped_column(DateTime)
    end_date: Mapped[datetime] = mapped_column(DateTime)
    available_tickets: Mapped[int] = mapped_column(Integer, index=True)

    def has_started(self):
        return self.start_date < datetime.now()

Теперь, когда мы добавили операции для взаимодействия с базой данных, давайте посмотрим, как мы можем использовать слой операций для работы с маршрутизацией:

@app.put("/tickets/{ticket_id}")
async def update_ticket(
    ticket_id: int, ticket_update: TicketUpdate, database:   Session = Depends(get_db)
) -> Ticket:
    try:
        return operations.update_ticket(ticket_id, ticket_update, database)
    except operations.NotFoundError:
        raise HTTPException(status_code=404)
    except operations.EventAlreadyStarted:
        raise HTTPException(status_code=400)
    
@app.delete("/tickets/{ticket_id}")
async def cancel_ticket(ticket_id: int, database: Session = Depends(get_db)) -> Ticket:
    try:
        return operations.delete_ticket(ticket_id, database)
    except operations.NotFoundError:
        raise HTTPException(status_code=404)
    except operations.EventAlreadyStarted:
        raise HTTPException(status_code=400)

Операционный уровень может быть использован для разделения обработки исключений при маршрутизации и взаимодействии с базой данных путем инкапсуляции операций с базой данных в отдельный модуль или класс. Этот модуль или класс может обрабатывать все взаимодействия с базой данных и вызывать пользовательские исключения для любых ошибок, возникающих при этих взаимодействиях.

Затем слой маршрутизации может импортировать этот модуль или класс и использовать его для выполнения необходимых операций с базой данных. Если в процессе работы с базой данных возникает ошибка, то пользовательское исключение, вызванное операционным уровнем, может быть поймано и обработано соответствующим образом уровнем маршрутизации, в нашем случае для ticketsNotFoundError и eventsEventAlreadyStarted, которые будут преобразованы в исключения маршрутизации HTTPException.

Такое разделение задач позволяет сделать код более чистым и удобным для сопровождения, поскольку уровень маршрутизации может сосредоточиться на обработке HTTP-запросов и ответов, а операционный уровень – на обработке взаимодействий с базой данных и ошибок.

Вот и все!