Open cmyui opened 7 months ago
An example from a recent PR:
class GrantType(StrEnum): IMPLICIT = "implicit" class ChannelMembership(TypedDict): session_id: str channel_name: str grant_type: GrantType created_at: datetime def serialize(channel_membership: ChannelMembership) -> str: """Serialize a channel membership to a string.""" serializable = { "session_id": channel_membership["session_id"], "channel_name": channel_membership["channel_name"], "grant_type": channel_membership["grant_type"], "created_at": channel_membership["created_at"].isoformat(), } return json.dumps(serializable) def deserialize(serialized: str) -> ChannelMembership: """Deserialize a channel membership from a string.""" deserialized = json.loads(serialized) return { "session_id": deserialized["session_id"], "channel_name": deserialized["channel_name"], "grant_type": GrantType(deserialized["grant_type"]), "created_at": datetime.fromisoformat(deserialized["created_at"]), }
And how they're used in other repo functions:
async def create( session_id: str, channel_name: str, grant_type: GrantType, ) -> ChannelMembership: """Create a new channel membership in redis.""" membership: ChannelMembership = { "session_id": session_id, "channel_name": channel_name, "grant_type": grant_type, "created_at": datetime.utcnow(), } serialized = serialize(membership) await app.state.services.redis.sadd( f"bancho:channel_memberships:{channel_name}", serialized, ) return membership async def fetch_all(channel_name: str) -> list[ChannelMembership]: """Fetch all channel memberships from redis.""" cursor = None channel_memberships = [] while cursor != 0: cursor, serialized_memberships = await app.state.services.redis.sscan( f"bancho:channel_memberships:{channel_name}", cursor=cursor or 0, ) for serialized in serialized_memberships: channel_membership = deserialize(serialized) channel_memberships.append(channel_membership) return channel_memberships
(Note this example serializes json data to/from strings, but it could be another format for any given technology we want to use in the repositories)
This will also supercede the @pymysql_encode decorators throughout the codebase
@pymysql_encode
any particular reason for doing this over doing this pydantic-side?
An example from a recent PR:
And how they're used in other repo functions:
(Note this example serializes json data to/from strings, but it could be another format for any given technology we want to use in the repositories)