Upgrade to Pro — share decks privately, control downloads, hide ads and more …

[PyconUS 2024] Having fun with Pydantic and pat...

[PyconUS 2024] Having fun with Pydantic and pattern matching

Pattern matching has been with us since Python3.10. Since the introduction of the match-case statement, we've got a powerful and elegant tool to control the flow of the program.

This talk is to showcase a real-world scenario of handling different messages coming from a broker, using match-case and Pydantic.

Sebastian Buczyński

May 18, 2024
Tweet

More Decks by Sebastian Buczyński

Other Decks in Technology

Transcript

  1. • Software Architect @ Sauce Labs • Trainer / Consultant

    @ Bottega IT Minds • I like memes • Blogger @ breadcrumbscollector.tech • Educate Pythonistas about software engineering whoami
  2. match latte: case Coffee() : print("It's caffee latte or latte

    machiato!") case Tea() : print("It's Matcha Latte") case _: print("Huh, no idea what it is”)
  3. match latte: case Coffee() : print("It's caffee latte or latte

    machiato!") case Tea() : print("It's Matcha Latte") case _: print("Huh, no idea what is it")
  4. match latte: case Coffee() : print("It's caffee latte or latte

    machiato!") case Tea() : print("It's Matcha Latte") case _: print("Huh, no idea what is it")
  5. price = 200 match price: case 100 : print("Cheap!") case

    200 : print("Expensive!") case _: print("No idea")
  6. price = 200 match price: case 100 : print("Cheap!") case

    200 : print("Expensive!") case _: print("No idea")
  7. price = 200 match price: case 100 : print("Cheap!") case

    200 : print("Expensive!") case _: print("No idea")
  8. price = 200 match price: case 100 : print("Cheap!") case

    200 : print("Expensive!") case _: print("No idea")
  9. price = 200 match price: case 100 : print("Cheap!") case

    200 : print("Expensive!") case _: print("No idea") What we’ll see? #1
  10. price = 2137 match price: case 100 : print("Cheap!") case

    200 : print("Expensive!") case _: print("No idea") What we’ll see? #2
  11. data = (1, 2, 3) match data: case {} :

    print("It's a dict!") case tuple() : print("Huh, a tuple!") case _: print("No idea") What we’ll see? #3
  12. data = {"name": "Sebastian"} match data: case {} : print("It's

    a dict!") case tuple() : print("Huh, a tuple!") case _: print("No idea") What we’ll see? #4
  13. What we’ll see? #5 data = {"name": "Sebastian"} match data:

    case {"name": _} : print("That's a dict with 'name'") case {} : print("It's a dict!") case _: print("No idea")
  14. What we’ll see? #6 data = {"name": "Sebastian"} match data:

    case {} : print("It's a dict!") case {"name": _} : print("That's a dict with 'name'") case _: print("No idea")
  15. What we’ll see? #6 data = {"name": "Sebastian"} match data:

    case {} : print("It's a dict!") case {"name": _} : print("That's a dict with 'name'") case _: print("No idea") Ordering of ‚cases’ is signi f icant
  16. Consuming stream of events: data {"type": "AccountCreated", "id": "64160ccb", "name":

    "Matthew"} {"type": "AccountUpdated", "id": "bb915f85", "old_status": "trial", "new_status": "paid"} {"type": "AccountDeleted", "id": "57ae55c5"}
  17. Consuming stream of events: data {"type": "AccountCreated", "id": "64160ccb", "name":

    "Matthew"} {"type": "AccountUpdated", "id": "bb915f85", "old_status": "trial", "new_status": "paid"} {"type": "AccountDeleted", "id": "57ae55c5"} {"type": "AccountUpdated", "id": "fede79b2", "old_status": "paid", "new_status": "trial"} {"type": "AccountCreated", "id": "23826148", "name": "John"}
  18. Consuming stream of events: data {"type": "AccountCreated", "id": "64160ccb", "name":

    "Matthew"} {"type": "AccountUpdated", "id": "bb915f85", "old_status": "trial", "new_status": "paid"} {"type": "AccountDeleted", "id": "57ae55c5"} {"type": "AccountUpdated", "id": "fede79b2", "old_status": "paid", "new_status": "trial"} {"type": "AccountCreated", "id": "23826148", "name": "John"} {"username": "Hecker"}
  19. def handle(payload: dict) -> None: # code like it's 2020

    😎 if payload.get("type") == "AccountCreated": ... elif payload.get("type") == "AccountDeleted": ... elif payload.get("type") == "AccountUpdated": ... else: print("Omg, what is this?!")
  20. def handle(payload: dict) -> None: # code like it's 2020

    😎 # ... and pretend we're in control 🙈🙉🙊 if payload.get("type") == "AccountCreated": ... elif payload.get("type") == "AccountDeleted": ... elif payload.get("type") == "AccountUpdated": if payload["new_status"] == "paid": ... elif payload["new_status"] == "trial": ... else: print("Omg, what is this?!")
  21. def handle(payload: dict) -> None: # HAMMER TIME! 🔨 match

    payload: case _: if payload.get("type") == "AccountCreated": ... elif payload.get("type") == "AccountDeleted": ... elif payload.get("type") == "AccountUpdated": if payload["new_status"] == "paid": ... elif payload["new_status"] == "trial": ... else: print("Omg, what is this?!")
  22. def handle(payload: dict) -> None: # HAMMER TIME! 🔨 match

    payload: case {"type": "AccountCreated"} : ... case _: if payload.get("type") == "AccountDeleted": ... elif payload.get("type") == "AccountUpdated": if payload["new_status"] == "paid": ... elif payload["new_status"] == "trial": ... else: print("Omg, what is this?!")
  23. def handle(payload: dict) -> None: # HAMMER TIME! 🔨 match

    payload: case {"type": "AccountCreated"} : ... case {"type": "AccountDeleted"} : ... case _: if payload.get("type") == "AccountUpdated": if payload["new_status"] == "paid": ... elif payload["new_status"] == "trial": ... else: print("Omg, what is this?!")
  24. def handle(payload: dict) -> None: # HAMMER TIME! 🔨 match

    payload: case {"type": "AccountCreated"} : ... case {"type": "AccountDeleted"} : ... case {"type": "AccountUpdated"} : if payload["new_status"] == "paid": ... elif payload["new_status"] == "trial": ... case _: print("Omg, what is this?!")
  25. def handle(payload: dict) -> None: # HAMMER TIME! 🔨 match

    payload: case {"type": "AccountCreated"} : ... case {"type": "AccountDeleted"} : ... case {"type": "AccountUpdated", "new_status": "paid"} : ... case {"type": "AccountUpdated", "new_status": "trial"} : ... case _: print("Omg, what is this?!")
  26. def handle(payload: dict) -> None: # HAMMER TIME! 🔨 match

    payload: case {"type": "AccountCreated"} : ... case {"type": "AccountDeleted"} : ... case {"type": "AccountUpdated", "new_status": "paid"} : ... case {"type": "AccountUpdated", "new_status": "trial"} : ... case _: print("Omg, what is this?!")
  27. Pydantic model data = {"type": "AccountCreated", "id": "64160ccb", "name": "Matthew"}

    class AccountCreated(BaseModel) : type: str id: str name: str AccountCreated( ** data) # AccountCreated(type='AccountCreated', id='64160ccb', name='Matthew')
  28. Pydantic model with a constant data = {"type": "AccountCreated", "id":

    "64160ccb", "name": "Matthew"} class AccountCreated(BaseModel) : type: Literal["AccountCreated"] id: str name: str AccountCreated(type="BAZINGA", id=„64160ccb", name="Matthew") # raises exception
  29. Pydantic models class AccountCreated(BaseModel) : ... class AccountDeleted(BaseModel) : ...

    class AccountUpdated(BaseModel) : type: Literal["AccountUpdated"] id: str old_status: Literal["trial", "paid"] new_status: Literal["trial", "paid"]
  30. def handle(payload: dict) -> None: match payload: case {"type": "AccountCreated"}

    : ... case {"type": "AccountDeleted"} : ... case {"type": "AccountUpdated", "new_status": "paid"} : ... case {"type": "AccountUpdated", "new_status": "trial"} : ... case _: print("Omg, what is this?!")
  31. def handle(payload: dict) -> None: match payload: case {"type": "AccountCreated"}

    : event = AccountCreated( ** payload) case {"type": "AccountDeleted"} : event = AccountDeleted( ** payload) case {"type": "AccountUpdated", "new_status": "paid"} : event = AccountUpdated( ** payload) case {"type": "AccountUpdated", "new_status": "trial"} : event = AccountUpdated( ** payload) case _: print("Omg, what is this?!”)
  32. def handle(payload: dict) -> None: supported_model = ( AccountCreated |

    AccountDeleted | AccountUpdated ) match payload: case {"type": "AccountCreated"} : ...
  33. def handle(payload: dict) -> None: supported_model = ( AccountCreated |

    AccountDeleted | AccountUpdated ) adapter = TypeAdapter(supported_model) match payload: case {"type": "AccountCreated"} : ...
  34. def handle(payload: dict) -> None: supported_model = ( AccountCreated |

    AccountDeleted | AccountUpdated ) adapter = TypeAdapter(supported_model) event = adapter.validate_python(payload) # event will be an instance of a supported model! match payload: case {"type": "AccountCreated"} : ...
  35. def handle(payload: dict) -> None: supported_model = ( AccountCreated |

    AccountDeleted | AccountUpdated ) adapter = TypeAdapter(supported_model) event = adapter.validate_python(payload) match event: case AccountCreated() : ... case AccountUpdated(new_status="paid") : ... case AccountUpdated(new_status="trial") : ...
  36. def handle(payload: dict) -> None: supported_model = ( AccountCreated |

    AccountDeleted | AccountUpdated | Any ) adapter = TypeAdapter(supported_model) event = adapter.validate_python(payload) match event: case AccountCreated() : ... case AccountUpdated(new_status="paid") : ... case AccountUpdated(new_status="trial") : ... case _: print("Omg, what is this?!")
  37. def handle(payload: dict) -> None: supported_model = ( AccountCreated |

    AccountDeleted | AccountUpdated | Any ) adapter = TypeAdapter(supported_model) event = adapter.validate_python(payload) match event: case AccountCreated() : ... case AccountUpdated(new_status="paid") : ... case AccountUpdated(new_status="trial") : ... case _: print("Omg, what is this?!")
  38. def handle(payload: dict) -> None: supported_model = ( AccountCreated |

    AccountDeleted | AccountUpdated | Any ) adapter = TypeAdapter(supported_model) event = adapter.validate_python(payload) match event: case AccountCreated() : ... case AccountUpdated(new_status="paid") : ... case AccountUpdated(new_status="trial") : ... case _: print("Omg, what is this?!")
  39. class AccountUpdatedToPaid(BaseModel) : type: Literal["AccountUpdated"] id: str old_status: Literal["trial"] new_status:

    Literal["paid"] class AccountUpdatedToTrial(BaseModel) : type: Literal["AccountUpdated"] id: str old_status: Literal["paid"] new_status: Literal["trial"]
  40. def handle(payload: dict) -> None: supported_model = ( ... |

    Any ) adapter = TypeAdapter(supported_model) event = adapter.validate_python(payload) match event: case AccountCreated() : ... case AccountUpdatedToPaid() : ... case AccountUpdatedToTrial() : ... case _: print("Omg, what is this?!")
  41. import functools @functools.singledispatch def event_handler(event: Any) -> None: print("Omg, what

    is this?!") @event_handler.register def handle_account_created(event: AccountCreated) -> None: ... @event_handler.register def handle_account_updated_to_paid(event: AccountUpdatedToPaid) -> None: ...
  42. ef handle(payload: dict) -> None: supported_model = ( AccountCreated |

    AccountDeleted | AccountUpdatedToTrial | AccountUpdatedToPaid | Any ) adapter = TypeAdapter(supported_model) event = adapter.validate_python(payload) event_handler(event)
  43. match latte: case Coffee() : print("It's caffee latte or latte

    machiato!") case Tea() : print("It's Matcha Latte") case _: print("Huh, no idea what is it") match…case on types
  44. match event: case AccountCreated() : ... case AccountUpdated(new_status="paid") : ...

    case AccountUpdated(new_status="trial") : ... case _: print("Omg, what is this?!") match…case on attributes' values
  45. pattern matching using Pydantic class AccountCreated(BaseModel) : type: Literal["AccountCreated"] id:

    str name: str class AccountDeleted(BaseModel) : type: Literal["AccountDeleted"] id: str supported_model = ( AccountCreated | AccountDeleted | AccountUpdated ) adapter = TypeAdapter(supported_model) event = adapter.validate_python(payload)
  46. pattern matching using Pydantic class AccountCreated(BaseModel) : type: Literal["AccountCreated"] id:

    str name: str class AccountDeleted(BaseModel) : type: Literal["AccountDeleted"] id: str supported_model = ( AccountCreated | AccountDeleted | AccountUpdated | Any ) adapter = TypeAdapter(supported_model) event = adapter.validate_python(payload)
  47. Dispatch based on type using @singledispatch import functools @functools.singledispatch def

    event_handler(event: Any) -> None: print("Omg, what is this?!") @event_handler.register def handle_account_created(event: AccountCreated) -> None: ...