Spaces:
Sleeping
Sleeping
File size: 5,048 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 136 137 138 139 140 |
from sqlalchemy.orm import sessionmaker
from models import Database_Entity
from repository import ConfigDatabase as cf
user_info = Database_Entity.UserInfo
users = Database_Entity.User
from sqlalchemy.orm import sessionmaker
import sys
import os
def getUserInfo(user_id: int) -> user_info:
try:
email = session.query(users.email).filter(users.id == user_id).one_or_none()
if email:
email = email[0]
user_record= session.query(user_info).filter(user_info.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 == user_id).one_or_none()
if email:
email = email[0]
user_record= session.query(user_info).filter(user_info.email == email).one_or_none()
if user_record:
session.close()
return user_record
else:
session.close()
return None
def getUserInfoByEmail(email:str) -> user_info:
try:
engine = cf.get_db_engine()
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
with Session() as session:
user_record= session.query(user_info).filter(user_info.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_info).filter(user_info.email == email).one_or_none()
if user_record:
session.close()
return user_record
else:
session.close()
return None
def addUserInfo(uid: str, email: str, display_name: str, photo_url: str) -> None:
try:
engine = cf.get_db_engine()
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
with Session() as session:
new_user = user_info(
uid = uid,
email = email,
display_name = display_name,
photo_url = photo_url
)
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_info(
uid = uid,
email = email,
display_name = display_name,
photo_url = photo_url
)
session.add(new_user)
session.commit()
session.close()
def updateUserInfo(user_id, uid: str, email: str, display_name: str, photo_url: str) -> None:
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 == user_id).one_or_none()
user_update= session.query(user_info).filter(user_info.email == email).one_or_none()
if user_update is not None:
user_update.uid = uid,
user_update.display_name = display_name,
user_update.photo_url = photo_url
session.commit()
session.close()
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 == user_id).one_or_none()
user_update= session.query(user_info).filter(user_info.email == email).one_or_none()
if user_update is not None:
user_update.uid = uid,
user_update.display_name = display_name,
user_update.photo_url = photo_url
session.commit()
session.close()
def updateImage(user_id, photo_url: str) -> None:
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 == user_id).one_or_none()
user_update= session.query(user_info).filter(user_info.email == email).one_or_none()
if user_update is not None:
user_update.photo_url = photo_url
session.commit()
session.close()
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 == user_id).one_or_none()
user_update = session.query(user_info).filter(user_info.email == email).one_or_none()
if user_update is not None:
user_update.photo_url = photo_url
session.commit()
session.close()
|