UUID v4 to v7

Author: Saiyashwanth7

pyproject.toml

@@ -14,6 +14,7 @@ dependencies = [
     "uvloop>=0.19.0",
     "httptools>=0.6.1",
     "uuid>=1.30",
+    "uuid6>=2024.1.12",
     "alembic>=1.13.1",
     "asyncpg>=0.29.0",
     "SQLAlchemy-Utils>=0.41.1",
@@ -117,4 +118,4 @@ explicit_package_bases = true
 
 [[tool.mypy.overrides]]
 module = "src.app.*"
-disallow_untyped_defs = true
+disallow_untyped_defs = true

src/app/api/v1/posts.py

@@ -59,7 +59,13 @@ async def read_posts(
     page: int = 1,
     items_per_page: int = 10,
 ) -> dict:
-    db_user = await crud_users.get(db=db, username=username, is_deleted=False, schema_to_select=UserRead)
+    db_user = await crud_users.get(
+        db=db, 
+        username=username, 
+        is_deleted=False, 
+        schema_to_select=UserRead,
+        return_as_model=True
+    )
     if not db_user:
         raise NotFoundException("User not found")
 
@@ -81,7 +87,13 @@ async def read_posts(
 async def read_post(
     request: Request, username: str, id: int, db: Annotated[AsyncSession, Depends(async_get_db)]
 ) -> PostRead:
-    db_user = await crud_users.get(db=db, username=username, is_deleted=False, schema_to_select=UserRead)
+    db_user = await crud_users.get(
+        db=db, 
+        username=username, 
+        is_deleted=False, 
+        schema_to_select=UserRead,
+        return_as_model=True
+    )
     if db_user is None:
         raise NotFoundException("User not found")
 
@@ -105,7 +117,13 @@ async def patch_post(
     current_user: Annotated[dict, Depends(get_current_user)],
     db: Annotated[AsyncSession, Depends(async_get_db)],
 ) -> dict[str, str]:
-    db_user = await crud_users.get(db=db, username=username, is_deleted=False, schema_to_select=UserRead)
+    db_user = await crud_users.get(
+        db=db, 
+        username=username, 
+        is_deleted=False, 
+        schema_to_select=UserRead,
+        return_as_model=True
+    )
     if db_user is None:
         raise NotFoundException("User not found")
 
@@ -130,7 +148,13 @@ async def erase_post(
     current_user: Annotated[dict, Depends(get_current_user)],
     db: Annotated[AsyncSession, Depends(async_get_db)],
 ) -> dict[str, str]:
-    db_user = await crud_users.get(db=db, username=username, is_deleted=False, schema_to_select=UserRead)
+    db_user = await crud_users.get(
+        db=db, 
+        username=username, 
+        is_deleted=False, 
+        schema_to_select=UserRead,
+        return_as_model=True
+    )
     if db_user is None:
         raise NotFoundException("User not found")
 
@@ -152,7 +176,13 @@ async def erase_post(
 async def erase_db_post(
     request: Request, username: str, id: int, db: Annotated[AsyncSession, Depends(async_get_db)]
 ) -> dict[str, str]:
-    db_user = await crud_users.get(db=db, username=username, is_deleted=False, schema_to_select=UserRead)
+    db_user = await crud_users.get(
+        db=db, 
+        username=username, 
+        is_deleted=False, 
+        schema_to_select=UserRead,
+        return_as_model=True
+    )
     if db_user is None:
         raise NotFoundException("User not found")
 
@@ -161,4 +191,4 @@ async def erase_db_post(
         raise NotFoundException("Post not found")
 
     await crud_posts.db_delete(db=db, id=id)
-    return {"message": "Post deleted from the database"}
+    return {"message": "Post deleted from the database"}

src/app/api/v1/users.py

@@ -85,15 +85,15 @@ async def patch_user(
         raise NotFoundException("User not found")
 
     db_user = cast(UserRead, db_user)
-    if db_user.username != current_user["username"]:
+    if db_user["username"] != current_user["username"]:
         raise ForbiddenException()
 
-    if values.username != db_user.username:
+    if values.username != db_user["username"]:
         existing_username = await crud_users.exists(db=db, username=values.username)
         if existing_username:
             raise DuplicateValueException("Username not available")
 
-    if values.email != db_user.email:
+    if values.email != db_user["email"]:
         existing_email = await crud_users.exists(db=db, email=values.email)
         if existing_email:
             raise DuplicateValueException("Email is already registered")

src/app/core/db/database.py

@@ -11,11 +11,13 @@ class Base(DeclarativeBase, MappedAsDataclass):
     pass
 
 
-DATABASE_URI = settings.POSTGRES_URI
+DATABASE_URI = settings.POSTGRES_URI 
 DATABASE_PREFIX = settings.POSTGRES_ASYNC_PREFIX
 DATABASE_URL = f"{DATABASE_PREFIX}{DATABASE_URI}"
 
-async_engine = create_async_engine(DATABASE_URL, echo=False, future=True)
+SQLITE_DB_URL='sqlite+aiosqlite:///./testing.db'
+
+async_engine = create_async_engine(SQLITE_DB_URL, echo=False, future=True)
 
 local_session = async_sessionmaker(bind=async_engine, class_=AsyncSession, expire_on_commit=False)
 

src/app/models/post.py

@@ -15,7 +15,7 @@ class Post(Base):
     created_by_user_id: Mapped[int] = mapped_column(ForeignKey("user.id"), index=True)
     title: Mapped[str] = mapped_column(String(30))
     text: Mapped[str] = mapped_column(String(63206))
-    uuid: Mapped[uuid_pkg.UUID] = mapped_column(default_factory=uuid7, primary_key=True, unique=True)
+    uuid: Mapped[uuid_pkg.UUID] = mapped_column(default_factory=uuid7, unique=True)
     media_url: Mapped[str | None] = mapped_column(String, default=None)
 
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default_factory=lambda: datetime.now(UTC))

src/app/models/user.py

@@ -1,8 +1,9 @@
-from uuid6 import uuid7 #126
+from uuid6 import uuid7
 from datetime import UTC, datetime
 import uuid as uuid_pkg
 
-from sqlalchemy import DateTime, ForeignKey, String,UUID
+from sqlalchemy import DateTime, ForeignKey, String
+from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.orm import Mapped, mapped_column
 
 from ..core.db.database import Base
@@ -11,19 +12,20 @@
 class User(Base):
     __tablename__ = "user"
 
-    id: Mapped[int] = mapped_column("id", autoincrement=True,primary_key=True, init=False)
-
+    # Option 1: Use integer ID as primary key (recommended for compatibility)
+    id: Mapped[int] = mapped_column(autoincrement=True, primary_key=True, init=False)
+    
     name: Mapped[str] = mapped_column(String(30))
     username: Mapped[str] = mapped_column(String(20), unique=True, index=True)
     email: Mapped[str] = mapped_column(String(50), unique=True, index=True)
     hashed_password: Mapped[str] = mapped_column(String)
 
     profile_image_url: Mapped[str] = mapped_column(String, default="https://profileimageurl.com")
-    uuid: Mapped[uuid_pkg.UUID] = mapped_column(UUID(as_uuid=True), default_factory=uuid7, primary_key=True)
+    uuid: Mapped[uuid_pkg.UUID] = mapped_column(UUID(as_uuid=True), default_factory=uuid7, unique=True)
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default_factory=lambda: datetime.now(UTC))
     updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None)
     deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None)
     is_deleted: Mapped[bool] = mapped_column(default=False, index=True)
     is_superuser: Mapped[bool] = mapped_column(default=False)
 
-    tier_id: Mapped[int | None] = mapped_column(ForeignKey("tier.id"), index=True, default=None, init=False)
+    tier_id: Mapped[int | None] = mapped_column(ForeignKey("tier.id"), index=True, default=None, init=False)

src/scripts/create_first_superuser.py

@@ -37,7 +37,7 @@ async def create_first_user(session: AsyncSession) -> None:
                 Column("email", String(50), nullable=False, unique=True, index=True),
                 Column("hashed_password", String, nullable=False),
                 Column("profile_image_url", String, default="https://profileimageurl.com"),
-                Column("uuid", UUID(as_uuid=True), primary_key=True, default=uuid7, unique=True),
+                Column("uuid", UUID(as_uuid=True), default=uuid7, unique=True),
                 Column("created_at", DateTime(timezone=True), default=lambda: datetime.now(UTC), nullable=False),
                 Column("updated_at", DateTime),
                 Column("deleted_at", DateTime),