Coverage for rfpy/jobs/dbcleanup.py: 0%

1import logging 

2from datetime import datetime, timezone 

3from sqlalchemy import or_ 

4from rfpy.model.humans import RefreshToken 

5 

6log = logging.getLogger(__name__) 

7 

8def cleanup_expired_tokens(session): 

9 """Remove expired and revoked refresh tokens from the database""" 

10 result = session.query(RefreshToken).filter( 

11 or_( 

12 RefreshToken.expires_at < datetime.now(timezone.utc), 

13 RefreshToken.revoked == True # noqa: E712 

14 ) 

15 ).delete(synchronize_session=False) 

16 

17 session.commit() 

18 log.info(f"Cleaned up {result} expired or revoked refresh tokens")