Lior Barak
06/23/2024, 2:58 PMCraig Harshbarger
06/24/2024, 3:19 PMLior Barak
06/24/2024, 3:24 PMfrom prefect import flow, task
from pydantic import BaseModel, Field, validator
class BaseDefineScan(WithCustomEncoder):
OrgId: str
SiteId: str
ScanId: str
class DefineScan(BaseDefineScan):
ShotsCount: Optional[int] = Field(default=200)
@validator('OrgId')
def check_org_is_ok(cls, org_id):
"""
some place holder code
"""
if org_id == "bad organization":
raise ValueError()
return org_id
@flow(task_runner="ConcurrentTaskRunner", persist_result=True, log_prints=True)
async def full_scan_flow(define_scan: DefineScan):
await all_scan_processing(define_scan)
Craig Harshbarger
06/24/2024, 3:31 PMLior Barak
06/24/2024, 3:43 PMfrom prefect import flow
from pydantic import BaseModel
class Model(BaseModel):
a: int
b: float
c: str
@flow
def model_validator(model: Model):
print(model)
this might have something to do with pydantic 1 vs 2.
i'm not sureLior Barak
06/25/2024, 12:08 PMfrom prefect import flow
from models import TestModel
@flow
def model_validator(model: TestModel):
print(model)
`models.py`:
from pydantic import BaseModel
class TestModel(BaseModel):
a: int
b: float
c: str