UPDATE: migrated one time util methods to blueprints

web_server/utils/user_utils.py

@@ -3,78 +3,13 @@ from typing import Optional, List
 from datetime import datetime
 from itsdangerous import URLSafeTimedSerializer, BadSignature, SignatureExpired
 from os import getenv
-from werkzeug.security import generate_password_hash, check_password_hash
+from werkzeug.security import generate_password_hash
 from dateutil import parser
 from dotenv import load_dotenv
 load_dotenv()
 
 serializer = URLSafeTimedSerializer(getenv("AUTH_SECRET_KEY"))
 
-def get_user_id(username: str) -> Optional[int]:
-    """
-    Returns user_id associated with given username
-    """
-    with Database() as db:
-        data = db.fetchone("""
-            SELECT user_id 
-            FROM users 
-            WHERE username = ?
-        """, (username,))
-    return data['user_id'] if data else None
-
-def get_username(user_id: str) -> Optional[str]:
-    """
-    Returns username associated with given user_id
-    """
-    with Database() as db:
-        data = db.fetchone("""
-            SELECT username 
-            FROM user 
-            WHERE user_id = ?
-        """, (user_id,))
-    return data['username'] if data else None
-
-def get_session_info_email(email: str) -> dict:
-    """
-    Returns username and user_id given email
-    """
-    with Database as db:
-        session_info = db.fetchone("""
-            SELECT user_id, username
-            FROM user
-            WHERE email = ?
-        """, (email,))
-        return session_info
-    
-def is_user_partner(user_id: int) -> bool:
-    """
-    Returns True if user is a partner, else False
-    """
-    with Database() as db:
-        data = db.fetchone("""
-            SELECT is_partnered 
-            FROM users 
-            WHERE user_id = ?
-        """, (user_id,))
-    return bool(data)
-
-def is_subscribed(user_id: int, subscribed_to_id: int) -> bool:
-    """
-    Returns True if user is subscribed to a streamer, else False
-    """
-    with Database() as db:
-        result = db.fetchone("""
-            SELECT *
-            FROM subscribes 
-            WHERE user_id = ? 
-            AND subscribed_id = ?
-            AND expires > ?;
-        """, (user_id, subscribed_to_id, datetime.now()))
-    print(result)
-    if result:
-        return True
-    return False
-
 def is_following(user_id: int, followed_id: int) -> bool:
     """
     Returns where a user is following another
@@ -116,27 +51,6 @@ def unfollow(user_id: int, followed_id: int):
         """, (user_id, followed_id))
     return {"success": True}
 
-
-def subscription_expiration(user_id: int, subscribed_id: int) -> int:
-    """
-    Returns the amount of time left until user subscription to a streamer ends
-    """
-    with Database() as db:
-        data = db.fetchone("""
-            SELECT expires 
-            FROM subscribes 
-            WHERE user_id = ? 
-            AND subscribed_id = ? 
-            AND expires > ?
-        """, (user_id, subscribed_id, datetime.now()))
-
-    if data:
-        expiration_date = data["expires"]
-        remaining_time = (parser.parse(expiration_date) - datetime.now()).seconds
-        return remaining_time
-
-    return 0
-
 def generate_token(email, salt_value) -> str:
     """
     Creates a token for password reset
@@ -174,35 +88,3 @@ def reset_password(new_password: str, email: str) -> bool:
     
     return True
 
-def get_email(user_id: int) -> Optional[str]:
-    with Database() as db:
-        email = db.fetchone("""
-            SELECT email
-            FROM users
-            WHERE user_id = ?
-        """, (user_id,))
-    
-    return email["email"] if email else None
-
-def get_followed_streamers(user_id: int) -> Optional[List[dict]]:
-    """
-    Returns a list of streamers who the user follows
-    """
-    with Database() as db:
-        followed_streamers = db.fetchall("""
-            SELECT user_id, username
-            FROM users
-            WHERE user_id IN (SELECT followed_id FROM follows WHERE user_id = ?);
-        """, (user_id,))
-    return followed_streamers
-
-def get_user(user_id: int) -> Optional[dict]:
-    """
-    Returns information about a user from user_id
-    """
-    with Database() as db:
-        data = db.fetchone("""
-            SELECT user_id, username, bio, num_followers, is_partnered, is_live FROM users
-            WHERE user_id = ?;
-        """, (user_id,))
-    return data