feat: implement unset

app/_typing.py

@@ -2,6 +2,26 @@
 
 from ipaddress import IPv4Address
 from ipaddress import IPv6Address
+from typing import Any
+from typing import TypeVar
 from typing import Union
 
 IPAddress = Union[IPv4Address, IPv6Address]
+T = TypeVar("T")
+
+
+class Unset:
+    def __repr__(self) -> str:
+        return "Unset"
+
+    def __copy__(self: T) -> T:
+        return self
+
+    def __reduce__(self) -> str:
+        return "Unset"
+
+    def __deepcopy__(self: T, _: Any) -> T:
+        return self
+
+
+UNSET = Unset()

app/repositories/ingame_logins.py

@@ -2,9 +2,13 @@
 
 import textwrap
 from typing import Any
+from typing import cast
 from typing import Optional
+from typing import TypedDict
 
 import app.state.services
+from app._typing import UNSET
+from app._typing import Unset
 
 # +--------------+------------------------+------+-----+---------+-------+
 # | Field        | Type                   | Null | Key | Default | Extra |
@@ -24,12 +28,28 @@
 )
 
 
+class InGameLogin(TypedDict):
+    id: int
+    userid: str
+    ip: str
+    osu_ver: str
+    osu_stream: str
+    datetime: datetime
+
+
+class InGameLoginUpdateFields(TypedDict, total=False):
+    userid: str
+    ip: str
+    osu_ver: str
+    osu_stream: str
+
+
 async def create(
     user_id: int,
     ip: str,
     osu_ver: str,
     osu_stream: str,
-) -> dict[str, Any]:
+) -> InGameLogin:
     """Create a new login entry in the database."""
     query = f"""\
         INSERT INTO ingame_logins (userid, ip, osu_ver, osu_stream, datetime)
@@ -51,9 +71,10 @@ async def create(
     params = {
         "id": rec_id,
     }
-    rec = await app.state.services.database.fetch_one(query, params)
-    assert rec is not None
-    return dict(rec)
+    ingame_login = await app.state.services.database.fetch_one(query, params)
+
+    assert ingame_login is not None
+    return cast(InGameLogin, ingame_login)
 
 
 async def fetch_one(
@@ -62,7 +83,7 @@ async def fetch_one(
     ip: Optional[str] = None,
     osu_ver: Optional[str] = None,
     osu_stream: Optional[str] = None,
-) -> Optional[dict[str, Any]]:
+) -> Optional[InGameLogin]:
     """Fetch a login entry from the database."""
     if (
         id is None
@@ -89,8 +110,9 @@ async def fetch_one(
         "osu_ver": osu_ver,
         "osu_stream": osu_stream,
     }
-    rec = await app.state.services.database.fetch_one(query, params)
-    return dict(rec) if rec is not None else None
+    ingame_login = await app.state.services.database.fetch_one(query, params)
+
+    return cast(InGameLogin, ingame_login) if ingame_login is not None else None
 
 
 async def fetch_count(
@@ -120,7 +142,7 @@ async def fetch_many(
     osu_stream: Optional[str] = None,
     page: Optional[int] = None,
     page_size: Optional[int] = None,
-) -> list[dict[str, Any]]:
+) -> list[InGameLogin]:
     """Fetch a list of logins from the database."""
     query = f"""\
         SELECT {READ_PARAMS}
@@ -145,34 +167,35 @@ async def fetch_many(
         params["limit"] = page_size
         params["offset"] = (page - 1) * page_size
 
-    recs = await app.state.services.database.fetch_all(query, params)
-    return [dict(rec) for rec in recs]
+    ingame_logins = await app.state.services.database.fetch_all(query, params)
+    return cast(list[InGameLogin], ingame_logins)
 
 
 async def update(
     id: int,
-    user_id: Optional[int] = None,
-    ip: Optional[str] = None,
-    osu_ver: Optional[str] = None,
-    osu_stream: Optional[str] = None,
+    user_id: int | Unset = UNSET,
+    ip: str | Unset = UNSET,
+    osu_ver: str | Unset = UNSET,
+    osu_stream: str | Unset = UNSET,
 ) -> Optional[dict[str, Any]]:
     """Update a login entry in the database."""
-    query = """\
+    update_fields = UserUpdateFields = {}
+    if not isinstance(user_id, Unset):
+        update_fields["user_id"] = user_id
+    if not isinstance(ip, Unset):
+        update_fields["ip"] = ip
+    if not isinstance(osu_ver, Unset):
+        update_fields["osu_ver"] = osu_ver
+    if not isinstance(osu_stream, Unset):
+        update_fields["osu_stream"] = osu_stream
+
+    query = f"""\
         UPDATE ingame_logins
-           SET userid = COALESCE(:userid, userid),
-               ip = COALESCE(:ip, ip),
-               osu_ver = COALESCE(:osu_ver, osu_ver),
-               osu_stream = COALESCE(:osu_stream, osu_stream)
+           SET {",".join(f"{k} = :{k}" for k in update_fields)}
          WHERE id = :id
     """
-    params = {
-        "id": id,
-        "userid": user_id,
-        "ip": ip,
-        "osu_ver": osu_ver,
-        "osu_stream": osu_stream,
-    }
-    await app.state.services.database.execute(query, params)
+    values = {"id": id} | update_fields
+    await app.state.services.database.execute(query, values)
 
     query = f"""\
         SELECT {READ_PARAMS}
@@ -182,8 +205,8 @@ async def update(
     params = {
         "id": id,
     }
-    rec = await app.state.services.database.fetch_one(query, params)
-    return dict(rec) if rec is not None else None
+    ingame_login = await app.state.services.database.fetch_one(query, params)
+    return cast(InGameLogin, ingame_login) if ingame_login is not None else None
 
 
 async def delete(id: int) -> Optional[dict[str, Any]]: