CHIEFSOFT\ameye
102 lines
2.8 KiB
Python
102 lines
2.8 KiB
Python
"""Application Models"""
|
|
import os
|
|
from dotenv import load_dotenv
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
import psycopg2
|
|
|
|
load_dotenv()
|
|
|
|
# DATABASE_URL=os.environ.get('DATABASE_URL') or 'mongodb://localhost:27017/myDatabase'
|
|
# print(DATABASE_URL)
|
|
# client = MongoClient(DATABASE_URL)
|
|
# db = client.myDatabase
|
|
|
|
dataUrl = os.getenv("DATABASE_URL")
|
|
db = psycopg2.connect(dataUrl)
|
|
|
|
|
|
class User:
|
|
"""User Model"""
|
|
def __init__(self):
|
|
return
|
|
|
|
def create(self, name="", email="", password=""):
|
|
"""Create a new user"""
|
|
user = self.get_by_email(email)
|
|
if user:
|
|
return
|
|
new_user = db.users.insert_one(
|
|
{
|
|
"name": name,
|
|
"email": email,
|
|
"password": self.encrypt_password(password),
|
|
"active": True
|
|
}
|
|
)
|
|
return self.get_by_id(new_user.inserted_id)
|
|
|
|
def get_all(self):
|
|
"""Get all users"""
|
|
users = db.users.find({"active": True})
|
|
return [{**user, "_id": str(user["_id"])} for user in users]
|
|
|
|
def get_by_id(self, user_id):
|
|
"""Get a user by id"""
|
|
user = db.users.find_one({"_id": bson.ObjectId(user_id), "active": True})
|
|
if not user:
|
|
return
|
|
user["_id"] = str(user["_id"])
|
|
user.pop("password")
|
|
return user
|
|
|
|
def get_by_email(self, email):
|
|
"""Get a user by email"""
|
|
user = db.users.find_one({"email": email, "active": True})
|
|
if not user:
|
|
return
|
|
user["_id"] = str(user["_id"])
|
|
return user
|
|
|
|
def update(self, user_id, name=""):
|
|
"""Update a user"""
|
|
data = {}
|
|
if name:
|
|
data["name"] = name
|
|
user = db.users.update_one(
|
|
{"_id": bson.ObjectId(user_id)},
|
|
{
|
|
"$set": data
|
|
}
|
|
)
|
|
user = self.get_by_id(user_id)
|
|
return user
|
|
|
|
def delete(self, user_id):
|
|
"""Delete a user"""
|
|
Books().delete_by_user_id(user_id)
|
|
user = db.users.delete_one({"_id": bson.ObjectId(user_id)})
|
|
user = self.get_by_id(user_id)
|
|
return user
|
|
|
|
def disable_account(self, user_id):
|
|
"""Disable a user account"""
|
|
user = db.users.update_one(
|
|
{"_id": bson.ObjectId(user_id)},
|
|
{"$set": {"active": False}}
|
|
)
|
|
user = self.get_by_id(user_id)
|
|
return user
|
|
|
|
def encrypt_password(self, password):
|
|
"""Encrypt password"""
|
|
return generate_password_hash(password)
|
|
|
|
def login(self, email, password):
|
|
"""Login a user"""
|
|
user = self.get_by_email(email)
|
|
print("THIS IS THE USER === > ", user)
|
|
if not user or not check_password_hash(user["password"], password):
|
|
return
|
|
user.pop("password")
|
|
return user
|