about summary refs log tree commit diff
diff options
context:
space:
mode:
authorFrederick Muriuki Muriithi2026-04-07 13:41:06 -0500
committerFrederick Muriuki Muriithi2026-04-07 13:54:01 -0500
commit135e861a8c09a93f4d094c8331529adadcefdba9 (patch)
tree5015cde2bdd99dececacb0b418ca994de8983568
parentb2e870ccb3cd82f8ba28a578144c20fa6a636221 (diff)
downloadgn-auth-135e861a8c09a93f4d094c8331529adadcefdba9.tar.gz
Add creator and creation time tracking to Resources.
-rw-r--r--gn_auth/auth/authentication/users.py4
-rw-r--r--gn_auth/auth/authorisation/resources/base.py52
2 files changed, 52 insertions, 4 deletions
diff --git a/gn_auth/auth/authentication/users.py b/gn_auth/auth/authentication/users.py
index 140ce36..fded79f 100644
--- a/gn_auth/auth/authentication/users.py
+++ b/gn_auth/auth/authentication/users.py
@@ -1,6 +1,6 @@
 """User-specific code and data structures."""
 import datetime
-from typing import Tuple
+from typing import Tuple, Union
 from uuid import UUID, uuid4
 from dataclasses import dataclass
 
@@ -26,7 +26,7 @@ class User:
         return self.user_id
 
     @staticmethod
-    def from_sqlite3_row(row: sqlite3.Row):
+    def from_sqlite3_row(row: Union[sqlite3.Row, dict]):
         """Generate a user from a row in an SQLite3 resultset"""
         return User(user_id=UUID(row["user_id"]),
                     email=row["email"],
diff --git a/gn_auth/auth/authorisation/resources/base.py b/gn_auth/auth/authorisation/resources/base.py
index 333ba0d..e4a1239 100644
--- a/gn_auth/auth/authorisation/resources/base.py
+++ b/gn_auth/auth/authorisation/resources/base.py
@@ -1,10 +1,17 @@
 """Base types for resources."""
+import logging
+import datetime
 from uuid import UUID
 from dataclasses import dataclass
-from typing import Any, Sequence
+from typing import Any, Sequence, Optional
 
 import sqlite3
 
+from gn_auth.auth.authentication.users import User
+
+
+logger = logging.getLogger(__name__)
+
 
 @dataclass(frozen=True)
 class ResourceCategory:
@@ -22,10 +29,49 @@ class Resource:
     resource_category: ResourceCategory
     public: bool
     resource_data: Sequence[dict[str, Any]] = tuple()
+    created_by: Optional[User] = None
+    created_at: datetime.datetime = datetime.datetime(1970, 1, 1, 0, 0, 0)
+
+    @staticmethod
+    def from_resource(# pylint: disable=[too-many-arguments, too-many-positional-arguments]
+            resource,
+            resource_id: Optional[UUID] = None,
+            resource_name: Optional[str] = None,
+            resource_category: Optional[ResourceCategory] = None,
+            public: Optional[bool] = None,
+            resource_data: Optional[Sequence[dict[str, Any]]] = None,
+            created_by: Optional[User] = None,
+            created_at: Optional[datetime.datetime] = None
+    ):
+        """Takes a Resource object `resource` and updates the attributes specified in `kwargs`."""
+        return Resource(
+            resource_id=resource_id or resource.resource_id,
+            resource_name=resource_name or resource.resource_name,
+            resource_category=resource_category or resource.resource_category,
+            public=bool(public) or resource.public,
+            resource_data=resource_data or resource.resource_data,
+            created_by=created_by or resource.created_by,
+            created_at=created_at or resource.created_at)
 
 
 def resource_from_dbrow(row: sqlite3.Row):
     """Convert an SQLite3 resultset row into a resource."""
+    try:
+        created_at = datetime.datetime.fromtimestamp(row["created_at"])
+    except IndexError as _ie:
+        created_at = datetime.datetime(1970, 1, 1, 0, 0, 0)
+
+    try:
+        created_by = User.from_sqlite3_row({
+            "user_id": row["creator_user_id"],
+            "email": row["creator_email"],
+            "name": row["creator_name"],
+            "verified": row["creator_verified"],
+            "created": row["creator_created"]
+        })
+    except IndexError as _ie:
+        created_by = None
+
     return Resource(
         resource_id=UUID(row["resource_id"]),
         resource_name=row["resource_name"],
@@ -33,4 +79,6 @@ def resource_from_dbrow(row: sqlite3.Row):
             UUID(row["resource_category_id"]),
             row["resource_category_key"],
             row["resource_category_description"]),
-        public=bool(int(row["public"])))
+        public=bool(int(row["public"])),
+        created_by=created_by,
+        created_at=created_at)