Make BeanieBaseUser and BeanieBaseAccessToken pure mixins to avoid duplicate collections

Author: frankie567

fastapi_users_db_beanie/__init__.py

@@ -1,5 +1,5 @@
 """FastAPI Users database adapter for Beanie."""
-from typing import TYPE_CHECKING, Any, Dict, Generic, Optional, Type, TypeVar
+from typing import Any, Dict, Generic, Optional, Type, TypeVar
 
 import bson.errors
 from beanie import Document, PydanticObjectId
@@ -13,9 +13,7 @@
 __version__ = "1.1.4"
 
 
-class BeanieBaseUser(Generic[ID], Document):
-    if TYPE_CHECKING:
-        id: ID  # type: ignore # pragma: no cover
+class BeanieBaseUser(BaseModel):
     email: str
     hashed_password: str
     is_active: bool = True
@@ -32,7 +30,11 @@ class Settings:
         ]
 
 
-UP_BEANIE = TypeVar("UP_BEANIE", bound=BeanieBaseUser)
+class BeanieBaseUserDocument(BeanieBaseUser, Document):  # type: ignore
+    pass
+
+
+UP_BEANIE = TypeVar("UP_BEANIE", bound=BeanieBaseUserDocument)
 
 
 class BaseOAuthAccount(BaseModel):
@@ -45,7 +47,9 @@ class BaseOAuthAccount(BaseModel):
     refresh_token: Optional[str] = None
 
 
-class BeanieUserDatabase(Generic[UP_BEANIE, ID], BaseUserDatabase[UP_BEANIE, ID]):
+class BeanieUserDatabase(
+    Generic[UP_BEANIE], BaseUserDatabase[UP_BEANIE, PydanticObjectId]
+):
     """
     Database adapter for Beanie.
 

fastapi_users_db_beanie/access_token.py

@@ -1,23 +1,33 @@
 from datetime import datetime, timezone
-from typing import Any, Dict, Generic, Optional, Type, TypeVar
+from typing import (
+    Any,
+    Dict,
+    Generic,
+    Optional,
+    Type,
+    TypeVar,
+)
 
-from beanie import Document
+from beanie import Document, PydanticObjectId
 from fastapi_users.authentication.strategy.db import AccessTokenDatabase
-from fastapi_users.models import ID
-from pydantic import Field
+from pydantic import BaseModel, Field
 from pymongo import IndexModel
 
 
-class BeanieBaseAccessToken(Generic[ID], Document):
+class BeanieBaseAccessToken(BaseModel):
     token: str
-    user_id: ID
+    user_id: PydanticObjectId
     created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
 
     class Settings:
         indexes = [IndexModel("token", unique=True)]
 
 
-AP_BEANIE = TypeVar("AP_BEANIE", bound=BeanieBaseAccessToken)
+class BeanieBaseAccessTokenDocument(BeanieBaseAccessToken, Document):  # type: ignore
+    pass
+
+
+AP_BEANIE = TypeVar("AP_BEANIE", bound=BeanieBaseAccessTokenDocument)
 
 
 class BeanieAccessTokenDatabase(Generic[AP_BEANIE], AccessTokenDatabase[AP_BEANIE]):

tests/test_access_token.py

@@ -3,7 +3,7 @@
 
 import pymongo.errors
 import pytest
-from beanie import PydanticObjectId, init_beanie
+from beanie import Document, PydanticObjectId, init_beanie
 from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
 
 from fastapi_users_db_beanie.access_token import (
@@ -12,7 +12,7 @@
 )
 
 
-class AccessToken(BeanieBaseAccessToken[PydanticObjectId]):
+class AccessToken(BeanieBaseAccessToken, Document):
     pass
 
 

tests/test_fastapi_users_db_beanie.py

@@ -2,7 +2,7 @@
 
 import pymongo.errors
 import pytest
-from beanie import PydanticObjectId, init_beanie
+from beanie import Document, PydanticObjectId, init_beanie
 from fastapi_users import InvalidID
 from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
 from pydantic import Field
@@ -15,7 +15,7 @@
 )
 
 
-class User(BeanieBaseUser[PydanticObjectId]):
+class User(Document, BeanieBaseUser):
     first_name: Optional[str] = None
 
 
@@ -70,7 +70,7 @@ async def beanie_user_db_oauth(
 
 @pytest.mark.asyncio
 async def test_queries(
-    beanie_user_db: BeanieUserDatabase[User, PydanticObjectId],
+    beanie_user_db: BeanieUserDatabase[User],
     oauth_account1: Dict[str, Any],
 ):
     user_create = {
@@ -140,7 +140,7 @@ async def test_queries(
     ],
 )
 async def test_email_query(
-    beanie_user_db: BeanieUserDatabase[User, PydanticObjectId],
+    beanie_user_db: BeanieUserDatabase[User],
     email: str,
     query: str,
     found: bool,
@@ -161,9 +161,7 @@ async def test_email_query(
 
 
 @pytest.mark.asyncio
-async def test_insert_existing_email(
-    beanie_user_db: BeanieUserDatabase[User, PydanticObjectId]
-):
+async def test_insert_existing_email(beanie_user_db: BeanieUserDatabase[User]):
     user_create = {
         "email": "lancelot@camelot.bt",
         "hashed_password": "guinevere",
@@ -176,7 +174,7 @@ async def test_insert_existing_email(
 
 @pytest.mark.asyncio
 async def test_queries_custom_fields(
-    beanie_user_db: BeanieUserDatabase[User, PydanticObjectId],
+    beanie_user_db: BeanieUserDatabase[User],
 ):
     """It should output custom fields in query result."""
     user_create = {
@@ -195,7 +193,7 @@ async def test_queries_custom_fields(
 
 @pytest.mark.asyncio
 async def test_queries_oauth(
-    beanie_user_db_oauth: BeanieUserDatabase[UserOAuth, PydanticObjectId],
+    beanie_user_db_oauth: BeanieUserDatabase[UserOAuth],
     oauth_account1: Dict[str, Any],
     oauth_account2: Dict[str, Any],
 ):