Spaces:
Sleeping
Sleeping
File size: 4,205 Bytes
ac3c54a d951fea ac3c54a d951fea ac3c54a d951fea ac3c54a d951fea ac3c54a d951fea ac3c54a d951fea ac3c54a d951fea ac3c54a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 |
import datetime
from bson import ObjectId
from streaksManagement import streaks_manager
def isexpired(previous_date):
# Get the current date and time
current_date =datetime.datetime.now()
# Convert the previous date (which is a string) to a datetime object
# Compare the two dates
if current_date > previous_date:
return True
else:
return False
def create_accessToken(db_uri: str, user_id: str, refresh_token: str) -> str:
from pymongo import MongoClient
current_time = datetime.datetime.now()
expire_at = current_time + datetime.timedelta(minutes=130)
"""
Inserts a new document into the specified MongoDB collection.
Parameters:
db_uri (str): MongoDB connection URI.
db_name (str): Name of the database.
collection_name (str): Name of the collection.
document (dict): The document to insert.
Returns:
str: The ID of the inserted document.
"""
# Connect to MongoDB
client = MongoClient(db_uri)
db = client["crayonics"]
collection = db["AccessToken"]
collection.find_one_and_delete({"refresh_token":refresh_token})
# Insert the document
result = collection.insert_one({"user_id":user_id,"refresh_token":refresh_token,"current_time":current_time,"expire_at":expire_at})
streaks_doc={}
streaks_doc['user_id'] = user_id
streaks_manager(db_uri=db_uri,document=streaks_doc)
client.close()
return str(result.inserted_id)
# Close the connection
def create_refreshToken(db_uri: str, user_id: str) -> str:
from pymongo import MongoClient
current_time = datetime.datetime.now()
expire_at = current_time + datetime.timedelta(days=30)
"""
Inserts a new document into the specified MongoDB collection.
Parameters:
db_uri (str): MongoDB connection URI.
user_id (str): id of user .
Returns:
str: The ID of the inserted document.
"""
# Connect to MongoDB
client = MongoClient(db_uri)
db = client["crayonics"]
collection = db["RefreshToken"]
# Insert the document
result = collection.insert_one({"user_id":user_id,"current_time":current_time,"expire_at":expire_at,"previous_access_token":"None"})
streaks_doc={}
streaks_doc['user_id'] = user_id
streaks_manager(db_uri=db_uri,document=streaks_doc)
client.close()
return str(result.inserted_id)
# Close the connection
def update_refreshTokenWithPreviouslyUsedAccessToken(db_uri: str, refresh_token: str,access_token:str) -> bool:
from pymongo import MongoClient
"""
"""
# Connect to MongoDB
client = MongoClient(db_uri)
db = client["crayonics"]
collection = db["RefreshToken"]
# Insert the document
try:
collection.update_one(
{"_id":ObjectId(oid=refresh_token) }, # Filter (find the document by user_id)
{"$set": {"previous_access_token": access_token}} # Add or update the field
)
client.close()
return True
except:
return False
def verify_access_token(db_uri: str, user_id: str, access_token: str) -> bool:
from pymongo import MongoClient
current_time = datetime.datetime.now()
expire_at = current_time + datetime.timedelta(minutes=15)
"""
"""
# Connect to MongoDB
client = MongoClient(db_uri)
db = client["crayonics"]
collection = db["AccessToken"]
doc = collection.find_one({"user_id":user_id})
if doc==None:
return False
else:
if str(doc['_id']) == access_token:
if isexpired(doc['expire_at']):
streaks_doc={}
streaks_doc['user_id'] = user_id
streaks_manager(db_uri=db_uri,document=streaks_doc)
return False
else:
streaks_doc={}
streaks_doc['user_id'] = user_id
streaks_manager(db_uri=db_uri,document=streaks_doc)
return True
else:
streaks_doc={}
streaks_doc['user_id'] = user_id
streaks_manager(db_uri=db_uri,document=streaks_doc)
return False
|