refresh tokens 0.1.2

src/database/auth/refresh_tokens.py

@@ -0,0 +1,63 @@
+
+from uuid import UUID
+from src.models.database_models.model import Model, engine, RefreshTokens
+from sqlalchemy import and_, not_, select
+from sqlalchemy.orm import sessionmaker
+from src.models.pydantic_models.model import RefreshTokensCreate, RefreshTokensOut
+class JwtCrudActions:
+    def __init__(self) -> None:
+        self.Session=sessionmaker(bind=engine)
+    
+    def get_token_by_user_id(self, user_id:UUID)->RefreshTokensOut|None:
+        with self.Session() as session:
+            with session.begin():
+                query=select(RefreshTokens).where(and_(RefreshTokens.user_id==user_id, not_(RefreshTokens.is_revoked)))
+                response=session.scalars(query).one_or_none()
+                if response is None:
+                    return None
+                return RefreshTokensOut.model_validate(response)
+            
+    def get_token_by_id(self, id:UUID)->RefreshTokensOut|None:
+        with self.Session() as session:
+            with session.begin():
+                query=select(RefreshTokens).where(RefreshTokens.id==id)
+                response=session.scalars(query).one_or_none()
+                if response is None:
+                    return None
+                return RefreshTokensOut.model_validate(response)
+    
+    def create_token(self, data:dict)->None:
+        with self.Session() as session:
+            with session.begin():
+                new_token=RefreshTokens(**data)
+                response=session.add(new_token)
+                return response
+            
+    def update_token(self, old_jti:UUID, new_jti:UUID)->bool:
+        with self.Session() as session:
+            with session.begin():
+                query=select(RefreshTokens).where(RefreshTokens.id==old_jti)
+                response=session.scalars(query).one()
+                response.is_revoked=True
+                response.replaced_by=new_jti
+                return True
+    
+    def revoke_all(self, user_id:UUID)->bool:
+        with self.Session() as session:
+            with session.begin():
+                query=select(RefreshTokens).where(RefreshTokens.user_id==user_id)
+                response=session.scalars(query).all()
+                for record in response:
+                    record.is_revoked=True
+                return True
+    
+    def logout(self,id:UUID)->bool:
+        with self.Session() as session:
+            with session.begin():
+                query=select(RefreshTokens).where(RefreshTokens.id == id)
+                response=session.scalars(query).one_or_none()
+                if response is None:
+                    return False
+                else:
+                    response.is_revoked=True
+                    return True