I’m writing a simple website that uses postgresql, sqlalchemy, and flask.
I have a server on which my DB is installed, and when I connect to it via this code: the connection is successful.
from sqlalchemy.orm import sessionmaker
from sshtunnel import SSHTunnelForwarder
from sqlalchemy import create_engine
from Back.App.helpers import singleton
from Back.App.orms import Base
from sqlalchemy.sql import and_
from datetime import date
import os
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
# wrapper is the same as:
# DbManager = singleton(DbManager)
# every db manager instance is the same to avoid multiple sessions, db connections and ssh tunnels
@singleton
class DbManager:
def __init__(self):
self.tunnel = None
self.engine = None
self.SessionFactory = None
self.session = None
self.start_tunnel()
def start_tunnel(self):
"""Start the SSH tunnel."""
self.tunnel = SSHTunnelForwarder(
(os.getenv('SSH_HOST'), int(os.getenv('SSH_PORT'))),
ssh_username=os.getenv('SSH_USERNAME'),
ssh_password=os.getenv('SSH_PASSWORD'),
remote_bind_address=(os.getenv('DB_HOST'), int(os.getenv('DB_PORT'))),
local_bind_address=('127.0.0.1', 6543)
)
self.tunnel.start()
print(f"SSH tunnel established on local port {self.tunnel.local_bind_port}")
# Set up the SQLAlchemy engine and session
db_url = (f"postgresql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@127.0.0.1:"
f"{self.tunnel.local_bind_port}/{os.getenv('DB_NAME')}")
self.engine = create_engine(db_url, pool_pre_ping=True)
Base.metadata.bind = self.engine
self.SessionFactory = sessionmaker(bind=self.engine)
self.session = self.SessionFactory()
def close(self):
"""Close the SSH tunnel and the SQLAlchemy session."""
if self.session:
self.session.close()
if self.tunnel:
self.tunnel.stop()
print("SSH tunnel closed.")
def get_objects(self, db_object, *filters):
"""
Fetch objects based on filters.
Returns a tuple: (success, data_or_message).
"""
try:
query = self.session.query(db_object)
if filters:
query = query.filter(and_(*filters))
objects = query.all()
if not objects:
return False, "No rows found in the database."
return True, objects
except Exception as e:
return False, f"Error fetching objects: {str(e)}"
from Back.App.orms import Student, Volunteer, AreaManager, Organization
from datetime import date
from db_connection import DbManager
db_manager = DbManager()
success, students = db_manager.get_objects(Student)
"students:"
for student in students:
print(f"ID: {student.student_id}, Name: {student.first_name} {student.last_name}, City: {student.city}")
I have an app.py file that currently has a basic login connection to the site, and when I try to open the first html page, I get an error that doesn’t allow me to see the screen.
the app.py file:
from flask import Flask, render_template, redirect, url_for, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_login import login_user, LoginManager, login_required, logout_user, current_user
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import InputRequired, Length
from flask_bcrypt import Bcrypt
from Back.App.orms import User, Student, Volunteer, AreaManager
from db_connection import DbManager
import os
# Initialize Flask app
app = Flask(__name__)
bcrypt = Bcrypt(app)
# Configure app settings
app.config['SQLALCHEMY_DATABASE_URI'] = (
f"postgresql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@"
f"127.0.0.1:6543/{os.getenv('DB_NAME')}")
app.config['SECRET_KEY'] = 'thisisasecretkey'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# Initialize database
db = SQLAlchemy(app)
# Set up Flask-Login
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'login'
@login_manager.user_loader
def load_user(user_id):
"""Load user by ID."""
return db.session.get(User, user_id)
# return db.session.query(User).get(user_id)
# Flask-WTF Forms
class RegisterForm(FlaskForm):
username = StringField(
validators=[InputRequired(), Length(min=4, max=20)],
render_kw={"placeholder": "id"}
)
password = PasswordField(
validators=[InputRequired(), Length(min=8, max=20)],
render_kw={"placeholder": "Password"}
)
submit = SubmitField('Register')
class LoginForm(FlaskForm):
username = StringField(
validators=[InputRequired(), Length(min=4, max=20)],
render_kw={"placeholder": "id"}
)
password = PasswordField(
validators=[InputRequired(), Length(min=8, max=20)],
render_kw={"placeholder": "Password"}
)
submit = SubmitField('Login')
# Flask routes
@app.route('/')
def home():
return render_template('home.html')
@app.route('/login', methods=['GET', 'POST'])
def login():
form = LoginForm()
if form.validate_on_submit():
db_manager = DbManager()
# Fetch the user
success, user_data = db_manager.get_objects(User, User.user_id == form.username.data)
# Check if the user exists
if not success:
return jsonify({"error": "Database error occurred while fetching user."}), 500
if not user_data:
return jsonify({"error": "Invalid username or password."}), 400
# Get the user object and verify the password
user = user_data[0]
if not bcrypt.check_password_hash(user.user_password, form.password.data):
return jsonify({"error": "Invalid username or password."}), 400
# Log in the user
login_user(user)
# Redirect based on user_type
if user.user_type == "student":
return redirect(url_for('student_home')), 200
elif user.user_type == "volunteer":
return redirect(url_for('volunteer_home')), 200
elif user.user_type == "area_manager":
return redirect(url_for('area_manager_home')), 200
else:
return jsonify({"error": "Invalid user type."}), 400
# Render the login form for GET requests or invalid submissions
return render_template('login.html', form=form), 200
@app.route('/student/home')
@login_required
def student_home():
return render_template('student_home.html')
@app.route('/volunteer/home')
@login_required
def volunteer_home():
return render_template('volunteer_home.html')
@app.route('/manager/home')
@login_required
def manager_home():
return render_template('area_manager_home.html')
@app.route('/logout')
@login_required
def logout():
logout_user()
return redirect(url_for('login'))
@app.route('/register', methods=['GET', 'POST'])
def register():
form = RegisterForm()
if form.validate_on_submit():
hashed_password = bcrypt.generate_password_hash(form.password.data).decode('utf-8')
db_manager = DbManager()
# Check if the username already exists
existing_username, user_list = db_manager.get_objects(User, User.user_id == form.username.data)
if existing_username:
return jsonify({"error": "id already exists. Please choose a different one."}), 400
# Check if the password hash already exists
existing_passwords, password_list = db_manager.get_objects(User, User.user_password == hashed_password)
print("user: ", User.user_password, "form: ", hashed_password)
print(existing_passwords)
if existing_passwords:
return jsonify({"error": "Password already exists. Please choose a different one."}), 400
# Determine user type
user_id = form.username.data
user_type = None
try:
if db_manager.get_objects(Student, Student.student_id == user_id):
user_type = "student"
elif db_manager.get_objects(Volunteer, Volunteer.volunteer_id == user_id):
user_type = "volunteer"
elif db_manager.get_objects(AreaManager, AreaManager.area_managers_id == user_id):
user_type = "area_manager"
else:
return jsonify({"error": "User ID not found in any table."}), 400
except Exception as e:
return jsonify({"error": f"An error occurred while verifying user ID: {str(e)}"}), 500
# Create the new user object
new_user = User(user_id=form.username.data, user_password=hashed_password, user_type=user_type)
success, message = db_manager.insert_object(new_user)
if success:
# return jsonify({"message": "User registered successfully."}), 201
return redirect(url_for('login')), 201
else:
return jsonify({"error": message}), 500
# return jsonify({"error": "Invalid form submission."}), 400
return render_template('register.html', form=form), 400
if __name__ == "__main__":
app.run(debug=True)
this is thy error: sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) connection to server at “127.0.0.1”, port 6543 failed: Connection refused (0x0000274D/10061)
Is the server running on that host and accepting TCP/IP connections?
and it fails on this line:
@app.route('/student/home')
@login_required
def student_home():
return render_template('student_home.html') #here it fails
The strange thing is that two days ago this connection via flask worked, and I didn’t change anything in the app.py file.
Does anyone have an idea what could be causing the problem?
i tried putting port 5432 and changing the host in the [‘SQLALCHEMY_DATABASE_URI’] in app.py, and i tried changing the sqlalchemy versions, with no success.
Dana Hartman is a new contributor to this site. Take care in asking for clarification, commenting, and answering.
Check out our Code of Conduct.