.

Author: pentest-gg

fastapi_users_db_dynamodb/access_token.py

@@ -22,12 +22,29 @@ class DynamoDBBaseAccessTokenTable(Model, Generic[ID]):
     __tablename__: str = config.get("DATABASE_TOKENTABLE_NAME")
 
     class Meta:
+        """The required `Meta` definitions for PynamoDB.
+
+        Args:
+            table_name (str): The name of the table.
+            region (str): The AWS region string where the table should be created.
+            billing_mode (str): The billing mode to use when creating the table. \
+            Currently only supports `PAY_PER_REQUEST`.
+        """
+
         table_name: str = config.get("DATABASE_TOKENTABLE_NAME")
         region: str = config.get("DATABASE_REGION")
         billing_mode: str = config.get("DATABASE_BILLING_MODE").value
 
     class CreatedAtIndex(GlobalSecondaryIndex):
+        """Enable the `created_at` attribute to be a Global Secondary Index.
+
+        Args:
+            GlobalSecondaryIndex (_type_): The Global Secondary Index base class.
+        """
+
         class Meta:
+            """The metadata for the Global Secondary Index."""
+
             index_name: str = "created_at-index"
             projection = AllProjection()
 
@@ -46,18 +63,29 @@ class Meta:
 
 
 class DynamoDBBaseAccessTokenTableUUID(DynamoDBBaseAccessTokenTable[UUID_ID]):
+    """A base class representing `AccessToken` objects with unique IDs.
+
+    Args:
+        DynamoDBBaseAccessTokenTable (_type_): The underlying table object.
+    """
+
     if TYPE_CHECKING:  # pragma: no cover
         user_id: UUID_ID
     else:
         user_id: GUID = GUID(null=False)
 
 
 class DynamoDBAccessTokenDatabase(Generic[AP], AccessTokenDatabase[AP]):
-    """Access token database adapter for AWS DynamoDB using aiopynamodb."""
+    """Access token database adapter for AWS DynamoDB using `aiopynamodb`."""
 
     access_token_table: type[AP]
 
     def __init__(self, access_token_table: type[AP]):
+        """Initialize the Database adapter.
+
+        Args:
+            access_token_table (type[AP]): The table for storing access tokens.
+        """
         self.access_token_table = access_token_table
 
     async def get_by_token(
@@ -66,7 +94,16 @@ async def get_by_token(
         max_age: datetime | None = None,
         instant_update: bool = False,
     ) -> AP | None:
-        """Retrieve an access token by token string."""
+        """Retrieve an access token by it's token string.
+
+        Args:
+            token (str): The actual token string to be looked for.
+            max_age (datetime | None, optional): The maximum age of an access token. Expired ones will not be returned. Defaults to None.
+            instant_update (bool, optional): Whether to use consistent reads. Defaults to False.
+
+        Returns:
+            AP | None: The access token, if found.
+        """
         await ensure_tables_exist(self.access_token_table)  # type: ignore
 
         try:
@@ -83,7 +120,18 @@ async def get_by_token(
             return None
 
     async def create(self, create_dict: dict[str, Any] | AP) -> AP:
-        """Create a new access token and return an instance of AP."""
+        """Create a new access token.
+
+        Args:
+            create_dict (dict[str, Any] | AP): A dictionary holding the data of the access token.
+
+        Raises:
+            ValueError: If the access token could not be created for whatever reason.
+            ValueError: If the access token could not be created because the table did not exist.
+
+        Returns:
+            AP: The newly created `AccessToken` object.
+        """
         await ensure_tables_exist(self.access_token_table)  # type: ignore
 
         if isinstance(create_dict, dict):
@@ -103,7 +151,19 @@ async def create(self, create_dict: dict[str, Any] | AP) -> AP:
         return token
 
     async def update(self, access_token: AP, update_dict: dict[str, Any]) -> AP:
-        """Update an existing access token."""
+        """Update an existing access token.
+
+        Args:
+            access_token (AP): The `AccessToken` object to be deleted.
+            update_dict (dict[str, Any]): A dictionary with the changes that should be applied.
+
+        Raises:
+            ValueError: If the access token could not be updated for whatever reason.
+            ValueError: If the access token could not be updated because the table did not exist.
+
+        Returns:
+            AP: The refreshed `AccessToken` object.
+        """
         await ensure_tables_exist(self.access_token_table)  # type: ignore
 
         try:
@@ -121,7 +181,15 @@ async def update(self, access_token: AP, update_dict: dict[str, Any]) -> AP:
             ) from e
 
     async def delete(self, access_token: AP) -> None:
-        """Delete an access token."""
+        """Delete an access token.
+
+        Args:
+            access_token (AP): The `AccessToken` object to be deleted.
+
+        Raises:
+            ValueError: If the access token could not be deleted for whatever reason.
+            ValueError: If the access token could not be deleted because the table did not exist.
+        """
         await ensure_tables_exist(self.access_token_table)  # type: ignore
 
         try:

fastapi_users_db_dynamodb/config.py

@@ -16,6 +16,7 @@ class BillingMode(StrEnum):
     Returns:
         _type_: The enum member, representing the billing mode.
     """
+
     PAY_PER_REQUEST = "PAY_PER_REQUEST"
     # PROVISIONED = "PROVISIONED"
 
@@ -34,6 +35,7 @@ class __ConfigMap(TypedDict):
     Args:
         TypedDict (_type_): The base type.
     """
+
     DATABASE_REGION: str
     # DATABASE_BILLING_MODE: BillingMode
     DATABASE_BILLING_MODE: Literal[BillingMode.PAY_PER_REQUEST]

tests/conftest.py

@@ -9,21 +9,25 @@
 
 class User(schemas.BaseUser):
     """A base class representing an `User`."""
+
     first_name: str | None
 
 
 class UserCreate(schemas.BaseUserCreate):
     """A base class representing the creation of an `User`."""
+
     first_name: str | None
 
 
 class UserUpdate(schemas.BaseUserUpdate):
     """A base class representing the update of an `User`."""
+
     pass
 
 
 class UserOAuth(User, schemas.BaseOAuthAccountMixin):
     """A base class representing an `User` with linked `OAuth` accounts."""
+
     pass
 
 

tests/test_access_token.py

@@ -24,6 +24,7 @@ class Base(Model):
     Args:
         Model (_type_): The PynamoDB base class definition.
     """
+
     pass
 
 
@@ -34,6 +35,7 @@ class AccessToken(DynamoDBBaseAccessTokenTableUUID, Base):
         DynamoDBBaseAccessTokenTableUUID (_type_): The underlying table object.
         Base (_type_): The PynamoDB base class definition.
     """
+
     __tablename__: str = config.get("DATABASE_TOKENTABLE_NAME") + "_test"
 
     class Meta:
@@ -45,6 +47,7 @@ class Meta:
             billing_mode (str): The billing mode to use when creating the table. \
             Currently only supports `PAY_PER_REQUEST`.
         """
+
         table_name: str = config.get("DATABASE_TOKENTABLE_NAME") + "_test"
         region: str = config.get("DATABASE_REGION")
         billing_mode: str = config.get("DATABASE_BILLING_MODE").value
@@ -57,6 +60,7 @@ class User(DynamoDBBaseUserTableUUID, Base):
         DynamoDBBaseUserTableUUID (_type_): The underlying table object.
         Base (_type_): The PynamoDB base class definition.
     """
+
     __tablename__: str = config.get("DATABASE_USERTABLE_NAME") + "_test"
 
     class Meta:
@@ -68,6 +72,7 @@ class Meta:
             billing_mode (str): The billing mode to use when creating the table. \
             Currently only supports `PAY_PER_REQUEST`.
         """
+
         table_name: str = config.get("DATABASE_USERTABLE_NAME") + "_test"
         region: str = config.get("DATABASE_REGION")
         billing_mode: str = config.get("DATABASE_BILLING_MODE").value
@@ -180,7 +185,7 @@ async def test_insert_existing_token(
     dynamodb_access_token_db: DynamoDBAccessTokenDatabase[AccessToken],
     user_id: UUID4,
 ):
-    """Test function that creates and saves an already existing `AccessToken`. 
+    """Test function that creates and saves an already existing `AccessToken`.
 
     Args:
         dynamodb_access_token_db (DynamoDBAccessTokenDatabase[AccessToken]): The database instance to use.

tests/test_misc.py

@@ -8,16 +8,19 @@
 
 class NotAModel:
     """A class representing an invalid `Model`."""
+
     pass
 
 
 class IncompleteModel(Model):
     """A class representing an incomplete `Model`, which misses required functions."""
+
     pass
 
 
 class ValidModel(Model):
     """A class representing a valid `Model`."""
+
     class Meta:
         """The required `Meta` definitions for PynamoDB.
 
@@ -27,6 +30,7 @@ class Meta:
             billing_mode (str): The billing mode to use when creating the table. \
             Currently only supports `PAY_PER_REQUEST`.
         """
+
         table_name: str = "valid_model_test"
         region: str = config.get("DATABASE_REGION")
         billing_mode: str = config.get("DATABASE_BILLING_MODE").value

tests/test_users.py

@@ -22,6 +22,7 @@ class Base(Model):
     Args:
         Model (_type_): The PynamoDB base class definition.
     """
+
     pass
 
 
@@ -32,6 +33,7 @@ class User(DynamoDBBaseUserTableUUID, Base):
         DynamoDBBaseUserTableUUID (_type_): The underlying table object.
         Base (_type_): The PynamoDB base class definition.
     """
+
     __tablename__: str = config.get("DATABASE_USERTABLE_NAME") + "_test"
 
     class Meta:
@@ -43,6 +45,7 @@ class Meta:
             billing_mode (str): The billing mode to use when creating the table. \
             Currently only supports `PAY_PER_REQUEST`.
         """
+
         table_name: str = config.get("DATABASE_USERTABLE_NAME") + "_test"
         region: str = config.get("DATABASE_REGION")
         billing_mode: str = config.get("DATABASE_BILLING_MODE").value
@@ -59,6 +62,7 @@ class OAuthBase(Model):
     Args:
         Model (_type_): The PynamoDB base class definition.
     """
+
     pass
 
 
@@ -69,6 +73,7 @@ class OAuthAccount(DynamoDBBaseOAuthAccountTableUUID, OAuthBase):
         DynamoDBBaseOAuthAccountTableUUID (_type_): The underlying table object.
         OAuthBase (_type_): The base class for `OAuth`.
     """
+
     pass
 
 
@@ -79,6 +84,7 @@ class UserOAuth(DynamoDBBaseUserTableUUID, OAuthBase):
         DynamoDBBaseUserTableUUID (_type_): The underlying table object.
         OAuthBase (_type_): The base class representing `OAuth` related models.
     """
+
     __tablename__: str = config.get("DATABASE_OAUTHTABLE_NAME") + "_test"
 
     class Meta:
@@ -90,6 +96,7 @@ class Meta:
             billing_mode (str): The billing mode to use when creating the table. \
             Currently only supports `PAY_PER_REQUEST`.
         """
+
         table_name: str = config.get("DATABASE_OAUTHTABLE_NAME") + "_test"
         region: str = config.get("DATABASE_REGION")
         billing_mode: str = config.get("DATABASE_BILLING_MODE").value