Merge pull request #49 from seapagan/add-timestamps

Author: seapagan

sqliter/model/model.py

@@ -38,6 +38,14 @@ class BaseDBModel(BaseModel):
     """
 
     pk: int = Field(0, description="The mandatory primary key of the table.")
+    created_at: int = Field(
+        default=0,
+        description="Unix timestamp when the record was created.",
+    )
+    updated_at: int = Field(
+        default=0,
+        description="Unix timestamp when the record was last updated.",
+    )
 
     model_config = ConfigDict(
         extra="ignore",

sqliter/sqliter.py

@@ -10,6 +10,7 @@
 
 import logging
 import sqlite3
+import time
 from typing import TYPE_CHECKING, Any, Optional, TypeVar, Union
 
 from typing_extensions import Self
@@ -444,6 +445,11 @@ def insert(self, model_instance: T) -> T:
         model_class = type(model_instance)
         table_name = model_class.get_table_name()
 
+        # Always set created_at and updated_at timestamps
+        current_timestamp = int(time.time())
+        model_instance.created_at = current_timestamp
+        model_instance.updated_at = current_timestamp
+
         # Get the data from the model
         data = model_instance.model_dump()
         # remove the primary key field if it exists, otherwise we'll get
@@ -530,6 +536,10 @@ def update(self, model_instance: BaseDBModel) -> None:
 
         primary_key = model_class.get_primary_key()
 
+        # Set updated_at timestamp
+        current_timestamp = int(time.time())
+        model_instance.updated_at = current_timestamp
+
         fields = ", ".join(
             f"{field} = ?"
             for field in model_class.model_fields

tests/test_context_manager.py

@@ -116,8 +116,8 @@ def test_commit_called_once_in_transaction(self, mocker, tmp_path) -> None:
         # Assert that the data was committed
         assert result is not None, "Data was not committed."
         assert (
-            result[1] == "test"
-        ), f"Expected slug to be 'test', but got {result[0]}"
+            result[3] == "test"
+        ), f"Expected slug to be 'test', but got {result[3]}"
 
         # Close the new connection
         new_conn.close()

tests/test_debug_logging.py

@@ -39,7 +39,8 @@ def test_debug_sql_output_basic_query(
 
         # Assert the SQL query was printed
         assert (
-            'Executing SQL: SELECT "pk", "name", "age", "is_active", "score", '
+            'Executing SQL: SELECT "pk", "created_at", "updated_at", "name", '
+            '"age", "is_active", "score", '
             '"nullable_field" FROM "complex_model" WHERE age = 30.5'
             in caplog.text
         )
@@ -55,7 +56,8 @@ def test_debug_sql_output_string_values(
 
         # Assert the SQL query was printed with the string properly quoted
         assert (
-            'Executing SQL: SELECT "pk", "name", "age", "is_active", "score", '
+            'Executing SQL: SELECT "pk", "created_at", "updated_at", "name", '
+            '"age", "is_active", "score", '
             '"nullable_field" FROM "complex_model" WHERE name = \'Alice\''
             in caplog.text
         )
@@ -71,7 +73,8 @@ def test_debug_sql_output_multiple_conditions(
 
         # Assert the SQL query was printed with multiple conditions
         assert (
-            'Executing SQL: SELECT "pk", "name", "age", "is_active", "score", '
+            'Executing SQL: SELECT "pk", "created_at", "updated_at", "name", '
+            '"age", "is_active", "score", '
             '"nullable_field" FROM "complex_model" WHERE name = \'Alice\' AND '
             "age = 30.5" in caplog.text
         )
@@ -87,7 +90,8 @@ def test_debug_sql_output_order_and_limit(
 
         # Assert the SQL query was printed with ORDER and LIMIT
         assert (
-            'Executing SQL: SELECT "pk", "name", "age", "is_active", "score", '
+            'Executing SQL: SELECT "pk", "created_at", "updated_at", "name", '
+            '"age", "is_active", "score", '
             '"nullable_field" FROM "complex_model" ORDER BY "age" DESC LIMIT 1'
             in caplog.text
         )
@@ -114,7 +118,8 @@ def test_debug_sql_output_with_null_value(
 
         # Assert the SQL query was printed with IS NULL
         assert (
-            'Executing SQL: SELECT "pk", "name", "age", "is_active", "score", '
+            'Executing SQL: SELECT "pk", "created_at", "updated_at", "name", '
+            '"age", "is_active", "score", '
             '"nullable_field" FROM "complex_model" WHERE age IS NULL'
             in caplog.text
         )
@@ -199,8 +204,8 @@ def test_manual_logger_respects_debug_flag(self, caplog) -> None:
 
         # Assert that log output was captured with the manually passed logger
         assert (
-            'Executing SQL: SELECT "pk", "name", "age", "is_active", "score", '
-            in caplog.text
+            'Executing SQL: SELECT "pk", "created_at", "updated_at", "name", '
+            '"age", "is_active", "score", ' in caplog.text
         )
 
     def test_manual_logger_above_debug_level(self, caplog) -> None:
@@ -228,7 +233,8 @@ def test_debug_sql_output_no_matching_records(
 
         # Assert that the SQL query was logged despite no matching records
         assert (
-            'Executing SQL: SELECT "pk", "name", "age", "is_active", "score", '
+            'Executing SQL: SELECT "pk", "created_at", "updated_at", "name", '
+            '"age", "is_active", "score", '
             '"nullable_field" FROM "complex_model" WHERE age = 100'
             in caplog.text
         )
@@ -242,7 +248,8 @@ def test_debug_sql_output_empty_query(
 
         # Assert that the SQL query was logged for a full table scan
         assert (
-            'Executing SQL: SELECT "pk", "name", "age", "is_active", "score", '
+            'Executing SQL: SELECT "pk", "created_at", "updated_at", "name", '
+            '"age", "is_active", "score", '
             '"nullable_field" FROM "complex_model"' in caplog.text
         )
 

tests/test_optional_fields.py

@@ -415,6 +415,8 @@ def test_exclude_all_fields_error(
                     "address",
                     "phone",
                     "occupation",
+                    "created_at",
+                    "updated_at",
                 ]
             ).fetch_all()
 

tests/test_query.py

@@ -561,27 +561,35 @@ def test_query_mixed_valid_invalid_filter(self, db_mock) -> None:
 
     def test_fetch_result_with_list_of_tuples(self, mocker) -> None:
         """Test _fetch_result when _execute_query returns list of tuples."""
+        # ensure we get a dependable timestamp
+        mocker.patch("time.time", return_value=1234567890)
+
         db = SqliterDB(memory=True)
 
         # Create some mock tuples (mimicking database rows)
         mock_result = [
-            ("1", "john", "John", "content"),
-            ("2", "jane", "Jane", "content"),
+            ("1", "1234567890", "1234567890", "john", "John", "content"),
+            ("2", "1234567890", "1234567890", "jane", "Jane", "content"),
         ]
 
         # Mock the _execute_query method on the QueryBuilder instance
         query = db.select(ExampleModel)
         mocker.patch.object(query, "_execute_query", return_value=mock_result)
 
-        # Perform the fetch (this will internally call _fetch_result)
+        # Perform the fetch_one (this will internally call _fetch_result)
         result = query.fetch_one()
 
         # Assert that the result is the first tuple in the list and correct type
         # and content
         assert not isinstance(result, list)
         assert isinstance(result, ExampleModel)
         assert result == ExampleModel(
-            pk=1, slug="john", name="John", content="content"
+            pk=1,
+            updated_at=1234567890,
+            created_at=1234567890,
+            slug="john",
+            name="John",
+            content="content",
         )
 
     def test_exclude_pk_raises_valueerror(self) -> None:

tests/test_sqliter.py

@@ -178,9 +178,9 @@ def test_insert_license(self, db_mock) -> None:
             cursor.execute("SELECT * FROM test_table WHERE slug = ?", ("mit",))
             result = cursor.fetchone()
             assert result[0] == 1
-            assert result[1] == "mit"
-            assert result[2] == "MIT License"
-            assert result[3] == "MIT License Content"
+            assert result[3] == "mit"
+            assert result[4] == "MIT License"
+            assert result[5] == "MIT License Content"
 
     def test_fetch_license(self, db_mock) -> None:
         """Test fetching a license by primary key."""
@@ -454,6 +454,8 @@ def test_select_with_exclude_all_fields_error(
             db_mock_detailed.select(
                 DetailedPersonModel,
                 exclude=[
+                    "created_at",
+                    "updated_at",
                     "name",
                     "age",
                     "email",
@@ -542,6 +544,8 @@ def test_complex_model_field_types(self, db_mock) -> None:
         # Expected types in SQLite (INTEGER, REAL, TEXT, etc.)
         expected_types = {
             "pk": "INTEGER",
+            "created_at": "INTEGER",
+            "updated_at": "INTEGER",
             "name": "TEXT",
             "age": "REAL",
             "price": "REAL",

tests/test_timestamps.py

@@ -0,0 +1,50 @@
+"""Class to test the `created_at` and `updated_at` timestamps."""
+
+from tests.conftest import ExampleModel
+
+
+class TestTimestamps:
+    """Test the `created_at` and `updated_at` timestamps."""
+
+    def test_insert_timestamps(self, db_mock, mocker) -> None:
+        """Test both timestamps are set on record insert."""
+        # Mock time.time() to return a fixed timestamp
+        mocker.patch("time.time", return_value=1234567890)
+
+        new_instance = ExampleModel(
+            slug="test", name="Test", content="Test content"
+        )
+
+        # Perform the insert operation
+        returned_instance = db_mock.insert(new_instance)
+
+        # Assert that both created_at and updated_at are set to the mocked
+        # timestamp
+        assert returned_instance.created_at == 1234567890
+        assert returned_instance.updated_at == 1234567890
+
+    def test_update_timestamps(self, db_mock, mocker) -> None:
+        """Test that the `updated_at` timestamp is updated on record update."""
+        # Mock time.time() to return a fixed timestamp for the update
+        mocker.patch("time.time", return_value=1234567890)
+
+        new_instance = ExampleModel(
+            slug="test", name="Test", content="Test content"
+        )
+
+        # Perform the insert operation
+        returned_instance = db_mock.insert(new_instance)
+
+        mocker.patch("time.time", return_value=1234567891)
+
+        # Perform the update operation
+        db_mock.update(returned_instance)
+
+        # Assert that created_at remains the same and updated_at changes to the
+        # new mocked value
+        assert (
+            returned_instance.created_at == 1234567890
+        )  # Should remain unchanged
+        assert (
+            returned_instance.updated_at == 1234567891
+        )  # Should be updated to the new timestamp