Spaces:
Running
Running
File size: 3,652 Bytes
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 |
import datetime
from bson import ObjectId
def is_current_date_greater_than_previous(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})
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"})
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 is_current_date_greater_than_previous(doc['expire_at']):
return False
else:
return True
else:
print("doc exists",str(doc['_id']),access_token,str(doc['_id']) == access_token )
return False
|