Spaces:
Sleeping
Sleeping
File size: 4,750 Bytes
13ba451 |
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 |
from sqlalchemy.orm import sessionmaker
from models import Database_Entity
from repository import ConfigDatabase as cf
user_login = Database_Entity.UserLogin
users = Database_Entity.User
def getUserLogin(email: str) -> user_login:
try:
engine = cf.get_db_engine()
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
with Session() as session:
user_record = session.query(user_login).filter(user_login.user_email == email).one_or_none()
if user_record:
session.close()
return user_record
else:
session.close()
return None
except:
engine = cf.get_db_engine1()
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
with Session() as session:
user_record = session.query(user_login).filter(user_login.user_email == email).one_or_none()
if user_record:
session.close()
return user_record
else:
session.close()
return None
def getUserLoginById(id: int) -> user_login:
try:
engine = cf.get_db_engine()
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
with Session() as session:
email = session.query(users.email).filter(users.id == id).one_or_none()[0]
user_record = session.query(user_login).filter(user_login.user_email == email).one_or_none()
if user_record:
session.close()
return user_record
else:
session.close()
return None
except:
engine = cf.get_db_engine1()
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
with Session() as session:
email = session.query(users.email).filter(users.id == id).one_or_none()[0]
user_record = session.query(user_login).filter(user_login.user_email == email).one_or_none()
if user_record:
session.close()
return user_record
else:
session.close()
return None
def addUserLogin(user_email: str, session_id : str) -> None:
try:
engine = cf.get_db_engine()
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
with Session() as session:
new_user = user_login(
user_email = user_email,
user_session_id = session_id
)
session.add(new_user)
session.commit()
session.close()
except:
engine = cf.get_db_engine1()
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
with Session() as session:
new_user = user_login(
user_email = user_email,
user_session_id = session_id
)
session.add(new_user)
session.commit()
session.close()
def updateUserLogin(email: str, session_id : str ) -> None:
try:
engine = cf.get_db_engine()
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
with Session() as session:
user_update= session.query(user_login).filter(user_login.user_email == email).one_or_none()
if user_update is not None:
user_update.user_session_id = session_id
session.commit()
session.close()
except:
engine = cf.get_db_engine1()
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
with Session() as session:
user_update= session.query(user_login).filter(user_login.user_email == email).one_or_none()
if user_update is not None:
user_update.user_session_id = session_id
session.commit()
session.close()
def getUserSessionIdByUserEmail(id: int) -> user_login:
try:
engine = cf.get_db_engine()
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
with Session() as session:
email = session.query(users.email).filter(users.id == id).one_or_none()[0]
session.commit()
user_record= session.query(user_login.user_session_id).filter(user_login.user_email == email).one_or_none()[0]
session.commit()
print(user_record)
if user_record:
session.close()
return user_record
else:
session.close()
return None
except:
engine = cf.get_db_engine1()
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
with Session() as session:
email = session.query(users.email).filter(users.id == id).one_or_none()[0]
session.commit()
user_record= session.query(user_login.user_session_id).filter(user_login.user_email == email).one_or_none()[0]
session.commit()
print(user_record)
if user_record:
session.close()
return user_record
else:
session.close()
return None |