94 lines
2.1 KiB
Python
94 lines
2.1 KiB
Python
from enum import Enum, StrEnum
|
|
from typing import Literal, NewType
|
|
|
|
from pydantic import BaseModel, JsonValue
|
|
|
|
from synctoy.state_machine import StateMachine
|
|
|
|
|
|
class Type(StrEnum):
|
|
Null = "null"
|
|
User = "user"
|
|
Message = "message"
|
|
|
|
ObjectId = NewType("ObjectId", str)
|
|
|
|
class RNull(BaseModel):
|
|
type: Literal[Type.Null] = Type.Null
|
|
id: ObjectId
|
|
|
|
def equivalent_to(self, other: StateMachine):
|
|
return self == other
|
|
|
|
def can_transition_from(self, old: StateMachine):
|
|
return True
|
|
|
|
def can_transition_to(self, new: StateMachine):
|
|
return True
|
|
|
|
class RUser(BaseModel):
|
|
type: Literal[Type.User] = Type.User
|
|
id: ObjectId
|
|
name: str
|
|
|
|
def equivalent_to(self, other: StateMachine):
|
|
return self == other
|
|
|
|
def can_transition_from(self, old: StateMachine):
|
|
if isinstance(old, RNull):
|
|
return True
|
|
return False
|
|
|
|
def can_transition_to(self, new: StateMachine):
|
|
return False
|
|
|
|
class RMessage(BaseModel):
|
|
type: Literal[Type.Message] = Type.Message
|
|
id: ObjectId
|
|
sender: ObjectId
|
|
content: str
|
|
|
|
def equivalent_to(self, other: StateMachine):
|
|
return self == other
|
|
|
|
def can_transition_from(self, old: StateMachine):
|
|
if isinstance(old, RNull):
|
|
return True
|
|
if isinstance(old, RMessage):
|
|
return True
|
|
return False
|
|
|
|
def can_transition_to(self, new: StateMachine):
|
|
if isinstance(new, RMessage):
|
|
return new.sender == self.sender
|
|
return True
|
|
|
|
NonNullRecord = RUser | RMessage
|
|
Record = RNull | NonNullRecord
|
|
|
|
|
|
|
|
def as_state_machine(r: Record) -> StateMachine:
|
|
return r
|
|
|
|
|
|
class ConditionType(StrEnum):
|
|
True_ = "true"
|
|
EquivalentTo = "equivalent_to"
|
|
|
|
|
|
class CTrue(BaseModel):
|
|
type: Literal[ConditionType.True_] = ConditionType.True_
|
|
|
|
def is_met(self, old: Record, new: Record):
|
|
return True
|
|
|
|
class CEquivalentTo(BaseModel):
|
|
type: Literal[ConditionType.EquivalentTo] = ConditionType.EquivalentTo
|
|
|
|
expected: Record
|
|
|
|
def is_met(self, old: Record, new: Record):
|
|
return old.equivalent_to(self.expected)
|
|
|
|
Condition = CTrue | CEquivalentTo |