add uv for project management and sqmodel to model entity

Author: darixsamani

alembic/env.py

@@ -13,6 +13,10 @@
 from asyncpg import Connection
 import asyncio
 
+from sqlmodel import SQLModel
+from alembic import context
+from models import *  # Import your SQLModel models here
+
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.
 config = context.config
@@ -37,6 +41,7 @@ def get_url():
 # for 'autogenerate' support
 # from myapp import mymodel
 # target_metadata = mymodel.Base.metadata
+target_metadata = SQLModel.metadata
 target_metadata = Base.metadata
 
 print(f"target_metadata: ")

database/dao/dao_posts.py

@@ -1,5 +1,5 @@
 from sqlalchemy.future import select
-from models.posts import Post
+from models.posts import Posts
 from schemas.posts import PostCreate
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -14,32 +14,32 @@ def __init__(self, db: AsyncSession):
 
 
     async def get_post_user(self, user_id: int):
-        result = await self.db.execute(select(Post).where(Post.user_id == user_id))
+        result = await self.db.execute(select(Posts).where(Posts.user_id == user_id))
         posts = result.scalars()
         return posts
 
     async def get_post_by_id(self, post_id: int):
-        result = await self.db.execute(select(Post).where(Post.id == post_id))
+        result = await self.db.execute(select(Posts).where(Posts.id == post_id))
         post = result.scalar_one_or_none()
         return post
 
     async def get_post_by_user_id(self, user_id:int):
 
-        result = await self.db.execute(select(Post).filter(Post.user_id == user_id))
+        result = await self.db.execute(select(Posts).filter(Posts.user_id == user_id))
         posts = result.scalars()
         return posts
 
 
 
     async def create_post(self, post_create:PostCreate, user_id: int):
-        post = Post(title=post_create.title, content=post_create.content, user_id=user_id)
+        post = Posts(title=post_create.title, content=post_create.content, user_id=user_id)
         self.db.add(post)
         await self.db.commit()
         await self.db.refresh(post)
 
         return post
 
-    async def delete_post(self, post: Post):
+    async def delete_post(self, post: Posts):
 
         await self.db.delete(post)
         await self.db.commit()

database/dao/dao_users.py

@@ -1,10 +1,9 @@
 from sqlalchemy.future import select
 from sqlalchemy.ext.asyncio import AsyncSession
-from models.users import User
+from models.users import Users
 from pydantic import EmailStr
 from passlib.context import CryptContext
 from schemas.users import UserCreate
-from models.users import User
 import asyncio
 
 hash_helper = CryptContext(schemes=["bcrypt"])
@@ -20,31 +19,31 @@ def __init__(self, db: AsyncSession) -> None:
         self.db = db
 
     async def get_users(self):
-        users = self.db.query(User).all()
+        users = self.db.query(Users).all()
         return users
 
 
     async def get_user_by_email(self, email: str):
 
-        result = await self.db.execute(select(User).where(User.email == email))
+        result = await self.db.execute(select(Users).where(Users.email == email))
         user = result.scalar_one_or_none()
         return user
 
     async def get_user_by_id(self, user_id: int):
-        result = await self.db.execute(select(User).where(User.id == user_id))
+        result = await self.db.execute(select(Users).where(Users.id == user_id))
         user = result.scalar_one_or_none()
         return user
 
 
     async def create_new_user(self, user: UserCreate):
-        user_db = User(email=user.email, fullname=user.fullname, password = hash_helper.encrypt(user.password))
+        user_db = Users(email=user.email, fullname=user.fullname, password = hash_helper.encrypt(user.password))
         self.db.add(user_db)
         await self.db.commit()
         await self.db.refresh(user_db)
         return user_db
 
     async def get_users(self, skip: int = 0, limit: int = 100):
-        result = await self.db.execute(select(User).offset(skip).limit(limit))
+        result = await self.db.execute(select(Users).offset(skip).limit(limit))
         users = result.scalars()
         return users
 

database/database.py

@@ -1,4 +1,5 @@
 import dotenv
+from sqlmodel import SQLModel
 from sqlalchemy import create_engine
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

models/__init__.py

@@ -1,2 +0,0 @@
-from .users import User
-from .posts import Post

models/posts.py

@@ -1,16 +1,13 @@
-from sqlalchemy import Column, Integer, String, ForeignKey
-from sqlalchemy.orm import Relationship
 from database.database import Base
-from sqlalchemy.orm import relationship
+from sqlmodel import Field, SQLModel
 
-class Post(Base):
 
-    __tablename__ = "posts"
+class Posts(SQLModel, table=True):
 
-    id: int = Column(Integer, primary_key=True, index=True)
-    title: str = Column(String,)
-    content: str = Column (String)
-    user_id: int = Column(Integer, ForeignKey('users.id'))
+    id: int = Field(default=None, primary_key=True)
+    title: str 
+    content: str 
+    user_id: int = Field(default=None, foreign_key='users.id')
 
     def __repr__(self):
         return f"<Post {self.id} {self.content[:20]}...>"

models/users.py

@@ -2,13 +2,13 @@
 from pydantic import EmailStr
 from sqlalchemy import Column, String, Integer
 from sqlalchemy.orm import relationship
+from sqlmodel import Field, SQLModel
 
-
-class User(Base):
-    __tablename__ = "users"
-    id: int = Column(Integer, primary_key=True, index=True)
-    email : EmailStr = Column(String(50), index=True, unique=True, nullable=False)
-    fullname : str = Column(String(50),)
+class Users(SQLModel, table = True):
+    
+    id: int = Field(default=None, primary_key=True)
+    email : EmailStr = Field(index=True, unique=True, nullable=False)
+    fullname : str 
     password: str = Column(String(100), nullable=False)
 
 

pyproject.toml

@@ -11,9 +11,11 @@ dependencies = [
     "dotenv>=0.9.9",
     "fastapi>=0.115.12",
     "passlib[bcrypt]>=1.7.4",
+    "psycopg2-binary>=2.9.10",
     "pydantic[email]>=2.11.7",
     "pyjwt>=2.10.1",
     "python-dotenv>=1.1.0",
     "python-multipart>=0.0.20",
+    "sqlmodel>=0.0.24",
     "uvicorn>=0.34.3",
 ]

routes/posts.py

@@ -2,7 +2,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from auth.deps import get_db, get_current_user
 from schemas.posts import PostCreate
-from models.users import User
+from models.users import Users
 from fastapi import HTTPException
 from database.dao.dao_posts import DaoPost
 from database.dao.dao_users import DaoUser
@@ -12,7 +12,7 @@
 
 
 @PostRouter.post("", response_model=PostResponseModel)
-async def create_new_post(post_create: PostCreate, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)):
+async def create_new_post(post_create: PostCreate, db: AsyncSession = Depends(get_db), user: Users = Depends(get_current_user)):
 
     daoPost = DaoPost(db=db)
     daoUser = DaoUser(db=db)
@@ -25,7 +25,7 @@ async def create_new_post(post_create: PostCreate, db: AsyncSession = Depends(ge
 
 
 @PostRouter.delete("/{id_post}")
-async def delete_posts(id_post:int,  db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)):
+async def delete_posts(id_post:int,  db: AsyncSession = Depends(get_db), user: Users = Depends(get_current_user)):
     daoPost = DaoPost(db=db)
 
     post_exist = await daoPost.get_post_by_id(post_id=id_post)

routes/users.py

@@ -1,6 +1,6 @@
 from fastapi import APIRouter, Depends, Body, status
 from schemas.users import UserCreate
-from models.users import User
+from models.users import Users
 from sqlalchemy.ext.asyncio import AsyncSession
 from database.dao.dao_posts import DaoPost
 from database.dao.dao_users import DaoUser
@@ -48,7 +48,7 @@ async def get_acces_token(db: AsyncSession = Depends(get_db), user_credentiel: O
 
 
 @UserRouter.get("/{user_id}/posts")
-async def get_all_posts_user(user_id:int, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user))-> List[PostResponseModel]:
+async def get_all_posts_user(user_id:int, db: AsyncSession = Depends(get_db), user: Users = Depends(get_current_user))-> List[PostResponseModel]:
 
     if user_id!=user.id:
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="your id does not match that of the authentication")
@@ -84,7 +84,7 @@ async def create_new_users(user: UserCreate, db: AsyncSession = Depends(get_db))
 
 
 @UserRouter.get("", response_model=List[UserResponseModel])
-async def get_all_user(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), skip: int = 0, limit: int = 100)->List[UserResponseModel]:
+async def get_all_user(db: AsyncSession = Depends(get_db), user: Users = Depends(get_current_user), skip: int = 0, limit: int = 100)->List[UserResponseModel]:
     daoUser = DaoUser(db=db)
     return await daoUser.get_users(skip=skip, limit=limit)