Added PK[T] annotations and update() without WHERE clause.

setup.py

@@ -3,13 +3,13 @@
 setup(
   name = 'targa',
   packages = ['targa', 'targa.errors'],
-  version = '1.0.0',
+  version = '1.0.2',
   license='MIT',
   description = 'A lightweight async Python library for MySQL queries and modeling.',
   author = 'whdev1',
   author_email = 'whdev1@protonmail.com',
   url = 'https://github.com/whdev1/targa',
-  download_url = 'https://github.com/whdev1/targa/archive/refs/tags/v1.0.0.tar.gz',
+  download_url = 'https://github.com/whdev1/targa/archive/refs/tags/v1.0.2.tar.gz',
   keywords = ['Targa', 'SQL', 'MySQL', 'async'],
   install_requires=['aiomysql'],
   classifiers=[

targa/__init__.py

@@ -1,2 +1,3 @@
 from .database import Database
+from .keys import PK
 from .model import Model

targa/database.py

@@ -1,3 +1,4 @@
+from .keys import _PK
 from .model import Model
 import aiomysql
 from .errors import InitializationError, MySQLErrors, SubstError
@@ -198,7 +199,7 @@ async def query(self, query: str, *substitutions, _ensure_conn: bool = True) ->
             ) for row in rows
         ]
 
-    async def update(self, model_inst: Model, where_clause: str) -> None:
+    async def update(self, model_inst: Model, where_clause: str = None) -> None:
         """
         Updates the specified model in the remote database using an UPDATE statement
         which includes the specified WHERE clause.
@@ -207,13 +208,32 @@ async def update(self, model_inst: Model, where_clause: str) -> None:
             model_inst: Model
                 The model instance that should be updated in the remote database.
 
-            where_clause: str
-                The WHERE clause to include in the SQL statement.
+            where_clause: str = None
+                The WHERE clause to include in the SQL statement. Optional if a primary key
+                was annotated in this model.
         
         Returns:
             Nothing
         """
 
+        # if no WHERE clause was provided, generate one based on a provided primary key annotation
+        if not where_clause:
+            # derive a list of annotation types
+            annotation_types: List[type] = [type(x[1]) for x in model_inst.__annotations__.items()]
+
+            # get the index of the first primary key annotation
+            pk_field_index: int = annotation_types.index(_PK)
+
+            # check that a primary key annotation was actually found
+            pk_field: str
+            if pk_field_index >= 0:
+                pk_field = list(model_inst.__annotations__.keys())[pk_field_index]
+            else:
+                raise KeyError('A WHERE clause is required if a primary key was not annotated')
+
+            # if it was, generate the WHERE clause
+            where_clause = f"WHERE {pk_field} = '{self._connection.escape_string(model_inst.__dict__[pk_field])}'"
+
         # ensure that a connection is established
         await self._ensure_connection()
 

targa/keys.py

@@ -0,0 +1,32 @@
+class _PK:
+    """
+    This class stands in for the _PK[T] annotation and contains
+    the type specified by the user.
+    """
+
+    _type: type = None
+
+    def __init__(self, _type: type) -> None:
+        """
+        Constructs a new _PK instance to be used in an annotation.
+        Not intended to be called directly; use a PK[T] annotation instead.
+
+        Parameters:
+            _type: type
+                The type that this key expects.
+        
+        Returns:
+            Nothing.
+        """
+
+        self._type = _type
+
+class _PK_Factory:
+    """
+    Factory for _PK instances. Provides the PK[T] annotation syntax.
+    """
+
+    def __getitem__(self, _type: type) -> _PK:
+        return _PK(_type)
+
+PK       = _PK_Factory()

targa/model.py

@@ -1,4 +1,5 @@
 from typing import Iterable, Optional
+from .keys import _PK
 
 class Model:
     def __init__(self, **kwargs) -> None:
@@ -22,18 +23,24 @@ def __init__(self, **kwargs) -> None:
 
         # loop over all of the fields defined in the derived class
         for field in self.__annotations__.keys():
+            # get the anticipated type of the field based on its annotation
+            expected_type = self.__annotations__[field]
+
             # ensure that the field was provided in the constructor and that it doesn't have
             # a default value already provided
             if field not in kwargs.keys() and field not in self.__dict__.keys():
                 raise AttributeError(
                     f"No value provided for field '{field}' of model {self.__class__.__name__}."
                 )
-
-            # type check the field based on its annotation
-            expected_type = self.__annotations__[field]
 
+            # check for a PK[T] annotation and unwrap one if necessary
+            if isinstance(expected_type, _PK):
+                # extract the type from the PK annotation
+                expected_type = expected_type._type
+
+            # check if the provided object is of the expected type
             if kwargs[field].__class__ != expected_type:
-                # check for an Optional[x] or Union[x, None] annotation
+                # check for an Optional[T] or Union[T, None] annotation
                 if hasattr(expected_type, '__args__') and expected_type.__args__[-1] == type(None):
                     # if this an Optional typing, check if the provided object is None. if not,
                     # it is invalid