64 lines
2.6 KiB
Python
64 lines
2.6 KiB
Python
|
|
from uuid import UUID
|
|
from src.models.database_models.model import engine, RefreshTokens
|
|
from sqlalchemy import and_, not_, select
|
|
from sqlalchemy.orm import sessionmaker
|
|
from src.models.pydantic_models.model import 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 |