Changeset View

Standalone View

class BaseRow:

TABLE: ClassVar[str]

PARTITION_KEY: ClassVar[Tuple[str, ...]]

CLUSTERING_KEY: ClassVar[Tuple[str, ...]] = ()

@classmethod

def from_dict(cls: Type[T], d: Dict[str, Any]) -> T:

return cls(**d) # type: ignore

olasdUnsubmitted

Not Done

def from_dict(cls: Type[T], d: Dict[str, Any]) -> T:

- return cls(**d) # type: ignore

+ cols = self.cols() # should probably be cached on the row class once it's been computed

+ return cls(**{k: v for (k, v) in d.items() if k in cols}) # type: ignore

@classmethod

olasd:

@classmethod

def cols(cls) -> List[str]:

return [field.name for field in dataclasses.fields(cls)]

def to_dict(self) -> Dict[str, Any]:

return dataclasses.asdict(self)

class OriginVisitStatusRow(BaseRow):

origin: str

visit: int

date: datetime.datetime

status: str

metadata: str

snapshot: bytes

@classmethod

def from_dict(cls: Type[T], d: Dict[str, Any]) -> T:

d = d.copy()

d.pop("type", None)

return cls(**d) # type: ignore

@dataclasses.dataclass

class OriginRow(BaseRow):

TABLE = "origin"

PARTITION_KEY = ("sha1",)

sha1: bytes

url: str