87 lines
3.5 KiB
Python
87 lines
3.5 KiB
Python
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import and_, not_, select
|
|
from sqlalchemy.exc import NoResultFound
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from src.errors.http_errors.errors import Errors
|
|
from src.models.database_models.model import RefreshTokens, engine
|
|
from src.models.pydantic_models.model import RefreshTokensOut
|
|
|
|
|
|
class JwtCrudActions:
|
|
def __init__(self) -> None:
|
|
self.Session=sessionmaker(bind=engine)
|
|
self.error=Errors()
|
|
|
|
def get_token_by_user_id(self, user_id:UUID)->RefreshTokensOut|None:
|
|
with self.Session() as session: # noqa: SIM117
|
|
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: # noqa: SIM117
|
|
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: # noqa: SIM117
|
|
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: # noqa: SIM117
|
|
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 create_and_update_token(self,data:dict, old_jti:UUID, new_jti:UUID)->bool:
|
|
with self.Session() as session: #noqa:SIM117
|
|
with session.begin():
|
|
new_token=RefreshTokens(**data)
|
|
session.add(new_token)
|
|
query=select(RefreshTokens).where(RefreshTokens.id==old_jti)
|
|
|
|
try:
|
|
response = session.scalars(query).one()
|
|
except NoResultFound as e:
|
|
raise self.error.not_found_error(detail="Token not found") from e
|
|
|
|
response.is_revoked=True
|
|
response.replaced_by=new_jti
|
|
return True
|
|
|
|
|
|
def revoke_all(self, user_id:UUID)->bool:
|
|
with self.Session() as session: # noqa: SIM117
|
|
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: # noqa: SIM117
|
|
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 |