mirror of
https://github.com/hatchet-dev/hatchet.git
synced 2026-01-05 16:19:43 -06:00
90805ca763
* feat: implement lifespans * feat: add psycopg test dep * feat: lifespan script * feat: examples and tests * drive-by: add license * drive-by: throw errors for sanity checks * feat: docs * feat: sync and async * fix: docs * fix: lint * fix: small tweaks * fix: rm unused * fix: typed example * fix: print
37 lines
674 B
Python
37 lines
674 B
Python
# ❓ Lifespan
|
|
|
|
from typing import AsyncGenerator, cast
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from hatchet_sdk import Context, EmptyModel, Hatchet
|
|
|
|
hatchet = Hatchet(debug=True)
|
|
|
|
|
|
class Lifespan(BaseModel):
|
|
foo: str
|
|
pi: float
|
|
|
|
|
|
async def lifespan() -> AsyncGenerator[Lifespan, None]:
|
|
yield Lifespan(foo="bar", pi=3.14)
|
|
|
|
|
|
@hatchet.task(name="LifespanWorkflow")
|
|
def lifespan_task(input: EmptyModel, ctx: Context) -> Lifespan:
|
|
return cast(Lifespan, ctx.lifespan)
|
|
|
|
|
|
def main() -> None:
|
|
worker = hatchet.worker(
|
|
"test-worker", slots=1, workflows=[lifespan_task], lifespan=lifespan
|
|
)
|
|
worker.start()
|
|
|
|
|
|
# ‼️
|
|
|
|
if __name__ == "__main__":
|
|
main()
|