Reputation: 19
I'm trying to avoid circular import at my Flask app (based on oracledb), but get "RuntimeError: Database pool not initialized". How can I properly handle the DB pool creation (create without using circular import)?
The project structure is:
.env
app.py
apps\__init__.py
db\__init__.py
logger\__init__.py
logger\forms.py
logger\util.py
...
The error is:
File "d:\project\apps\home\routes.py", line 14, in contacts
from apps.logger.util import log_feedback
File "d:\project\apps\logger\util.py", line 7, in <module>
db_pool = DatabasePool.get_pool()
File "d:\project\apps\db\__init__.py", line 25, in get_pool
raise RuntimeError("Database pool not initialized")
RuntimeError: Database pool not initialized
# .env
FLASK_APP=app.py
FLASK_ENV=development
FLASK_DEBUG=1
# Server configuration
HOST=0.0.0.0
PORT=5000
# Security
SECRET_KEY=your-secret-key-here
SECURITY_PASSWORD_SALT=fkslkfsdlkfnsdfnsfd
# Database configuration
DB_USER=test
DB_PASSWORD=password
DB_DSN=localhost:1521/FREEPDB1
# app.py
from apps import create_app
import os
app = create_app()
if __name__ == '__main__':
app.run(
host=os.getenv('HOST', '0.0.0.0'),
port=int(os.getenv('PORT', 5000)),
debug=os.getenv('FLASK_DEBUG', True),
)
# apps.__init__.py
from importlib import import_module
from flask import Flask
from dotenv import load_dotenv
from flask_login import LoginManager
import os
from datetime import timedelta
from apps.db import DatabasePool
login_manager = LoginManager()
def register_blueprints(app):
for module_name in ('home', 'Turnstile', 'authentication', 'dashboard'):
module = import_module('apps.{}.routes'.format(module_name))
app.register_blueprint(module.blueprint)
def create_app():
app = Flask(__name__,
template_folder='templates')
# Load environment variables
load_dotenv()
# Verify required environment variables are present
required_vars = ['DB_USER', 'DB_PASSWORD', 'DB_DSN', 'SECRET_KEY', 'SECURITY_PASSWORD_SALT']
missing_vars = [var for var in required_vars if not os.getenv(var)]
if missing_vars:
raise RuntimeError(f"Missing required environment variables: {', '.join(missing_vars)}")
# Configure app from environment variables
app.config.update(
SECRET_KEY=os.getenv('SECRET_KEY'),
SECURITY_PASSWORD_SALT=os.getenv('SECURITY_PASSWORD_SALT'),
DB_USER=os.getenv('DB_USER'),
DB_PASSWORD=os.getenv('DB_PASSWORD'),
DB_DSN=os.getenv('DB_DSN'),
SESSION_PROTECTION='strong',
PERMANENT_SESSION_LIFETIME = timedelta(minutes=60)
# Add other configuration settings as needed
)
# Initialize DB pool after config is loaded
with app.app_context():
DatabasePool.initialize(app)
# Register teardown callback
@app.teardown_appcontext
def close_db_pool(exception=None):
DatabasePool.close()
# Initialize Flask-Login
login_manager.init_app(app)
login_manager.login_view = 'authentication_blueprint.login'
login_manager.login_message = 'Please log in to access this page.'
# User loader callback
@login_manager.user_loader
def load_user(user_id):
from apps.models.models import User
return User.get_by_id(int(user_id))
# Register blueprints
register_blueprints(app)
return app
# apps.db.__init__.py
import oracledb
class DatabasePool:
_instance = None
_pool = None
@classmethod
def initialize(cls, app):
"""Initialize the database pool with app config"""
if cls._pool is None:
with app.app_context():
cls._pool = oracledb.create_pool(
user=app.config['DB_USER'],
password=app.config['DB_PASSWORD'],
dsn=app.config['DB_DSN'],
min=2,
max=5,
increment=1
)
@classmethod
def get_pool(cls):
"""Get the database pool instance"""
if cls._pool is None:
raise RuntimeError("Database pool not initialized")
return cls._pool
@classmethod
def close(cls):
"""Close the database pool"""
if cls._pool is not None:
cls._pool.close()
cls._pool = None
# apps.logger.util.py
from datetime import datetime
from apps.db import DatabasePool
from oracledb import DatabaseError
from apps.logger import FeedbackCategory
db_pool = DatabasePool.get_pool()
I can't initialize my DB within apps._init_.py becase there are blueprint imports as well so this approach cause circular import. So I have decided to move DB init into the separate module and import the DB pool in the app init module and other sub-modules, but this doesn't work as expected even with a wrapper function as below:
from apps.db import DatabasePool
def get_db_pool():
return DatabasePool.get_pool()
Upvotes: 0
Views: 13