user2312787
user2312787

Reputation: 19

Flask app DB module avoiding circular import

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

Answers (0)

Related Questions