Мобильная версияНиже приведено содержимое моего скрипта авторизации, содержащее определение модели пользователя, методы регистрации и входа, а также вспомогательный метод check_login:
auth_blueprint = Blueprint('auth', __name__, template_folder='templates')
# Setup Flask-Login and Bcrypt
bcrypt = Bcrypt()
login_manager = LoginManager()
login_manager.login_view = "auth.login"
# MongoDB configuration
mongo_uri = config.mongo_uri
client = MongoClient(mongo_uri)
db = client[config.DB_NAME]
users_collection = db["users"]
# User class for Flask-Login
class User(UserMixin):
def __init__(self, id, username, password):
self.id = id
self.username = username
self.password = password
@staticmethod
def get(user_id):
user_data = users_collection.find_one({"_id": user_id})
if user_data:
return User(str(user_data["_id"]), user_data["username"], user_data["password"])
return None
def get_id(self):
return self.id
@login_manager.user_loader
def load_user(user_id):
print(f'load_user method is called and the user_id is {user_id}')
try:
return User.get(user_id)
except:
return None
@auth_blueprint.route("/register", methods=["GET", "POST"])
def register():
if request.method == "POST":
username = request.form["username"]
password = request.form["password"]
confirmed_password = request.form["repeat-password"]
if not username:
flash("Please enter a username.", "danger")
return redirect(url_for("auth.register"))
if users_collection.find_one({"username": username}):
flash("Username already exists!", "danger")
return redirect(url_for("auth.register"))
if not password:
flash("Please enter a password.", "danger")
return redirect(url_for("auth.register"))
if not confirmed_password:
flash("Please confirm the password.", "danger")
return redirect(url_for("auth.register"))
if password != confirmed_password:
flash("Passwords must match!", "danger")
return redirect(url_for("auth.register"))
hashed_password = bcrypt.generate_password_hash(password).decode("utf-8")
users_collection.insert_one({"username": username, "password": hashed_password})
flash("Registration successful!", "success")
return redirect(url_for("auth.login"))
return render_template("register.html")
@auth_blueprint.route("/login", methods=["GET", "POST"])
def login():
if request.method == "POST":
username = request.form["username"]
password = request.form["password"]
# Check if the user exists and credentials are valid
user_data = users_collection.find_one({"username": username})
print(user_data)
if user_data and bcrypt.check_password_hash(user_data["password"], password):
session.clear()
user = User(str(user_data['_id']), user_data['username'], user_data['password'])
print(f'User id is {user.id}')
result = login_user(user, remember=True, duration=timedelta(days=7), force=True)
session.modified = True
print(f'Is user authenticated in login function: {current_user._get_current_object().is_authenticated}')
# print(f'Current user type is {type(current_user.is_authenticated)}')
if result:
print("Loggin successful!")
#print(f"Session after login: {session}") # Inspect session object
flash("Login successful!", "success")
return redirect(url_for("index"))
else:
print("Something went wrong
")return redirect(url_for("auth.login"))
# session['username'] = user.username
# if not helpers.url_has_allowed_host_and_scheme(next, request.host):
# return abort(400)
else:
flash("Invalid credentials!", "danger")
return render_template("login.html")
@auth_blueprint.route("/check_login")
def check_login():
print(f'Is user authenticated in check_login function: {current_user._get_current_object().is_authenticated}')
print(type(current_user._get_current_object()))
if current_user._get_current_object().is_authenticated:
return f"User {current_user.name} is logged in."
else:
return "No user is logged in."
P.S. Часть HTML, которая должна меняться в зависимости от входа пользователя в систему, также зависит от атрибута current_user._get_current_object().is_authenticated , поэтому я опустил ее, поскольку она показалась избыточной. Я также хотел бы отметить, что метод _get_current_object() прокси current_user ничего не меняет, и я использую его как взаимозаменяемый в целях тестирования.
Я просмотрел другие подобные вопросы и официальную документацию, но безуспешно. Мне кажется, что проблема сводится к тому, что прокси current_user не запоминается правильно, когда я перенаправляю его из функции входа; отладка во время функции входа в систему показывает, что для параметра current_user установлен экземпляр моего класса User , а для атрибута is_authenticated установлено значение True, но один раз Я перенаправляю в функцию check_login() прокси-сервер current_user по умолчанию на объект типа AnonymousUserMixin. Я подозреваю, что что-то не так с обработкой свойств сеанса , но я не могу этого понять.
EDIT:
Проблема была в функции load_user, я пытался получить идентификатор объекта без конвертации в формат BSON, который использует MongoDB. Исправленный метод показан ниже. Если оставить этот вопрос, это может помочь кому-то решить аналогичную тривиальную проблему.
@login_manager.user_loader
def load_user(user_id):
user_data = users_collection.find_one({"_id": ObjectId(user_id)})
if user_data:
return User(str(user_data['_id']), user_data['username'], user_data['password'])
return None
Подробнее здесь: https://stackoverflow.com/questions/793 ... -in-python
