refactor: assembling the app features in modular way (#9129)

Signed-off-by: -LAN- <laipz8200@outlook.com>
Co-authored-by: -LAN- <laipz8200@outlook.com>
This commit is contained in:
Bowen Liang 2024-11-30 23:05:22 +08:00 committed by GitHub
parent 3bc4dc58d7
commit 9b46b02717
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
30 changed files with 396 additions and 318 deletions

View File

@ -48,6 +48,8 @@ jobs:
cp .env.example .env
- name: Run DB Migration
env:
DEBUG: true
run: |
cd api
poetry run python -m flask upgrade-db

View File

@ -1,113 +1,13 @@
import os
import sys
python_version = sys.version_info
if not ((3, 11) <= python_version < (3, 13)):
print(f"Python 3.11 or 3.12 is required, current version is {python_version.major}.{python_version.minor}")
raise SystemExit(1)
from configs import dify_config
if not dify_config.DEBUG:
from gevent import monkey
monkey.patch_all()
import grpc.experimental.gevent
grpc.experimental.gevent.init_gevent()
import json
import threading
import time
import warnings
from flask import Response
from app_factory import create_app
from libs import threadings_utils, version_utils
# DO NOT REMOVE BELOW
from events import event_handlers # noqa: F401
from extensions.ext_database import db
# TODO: Find a way to avoid importing models here
from models import account, dataset, model, source, task, tool, tools, web # noqa: F401
# DO NOT REMOVE ABOVE
warnings.simplefilter("ignore", ResourceWarning)
os.environ["TZ"] = "UTC"
# windows platform not support tzset
if hasattr(time, "tzset"):
time.tzset()
# preparation before creating app
version_utils.check_supported_python_version()
threadings_utils.apply_gevent_threading_patch()
# create app
app = create_app()
celery = app.extensions["celery"]
if dify_config.TESTING:
print("App is running in TESTING mode")
@app.after_request
def after_request(response):
"""Add Version headers to the response."""
response.headers.add("X-Version", dify_config.CURRENT_VERSION)
response.headers.add("X-Env", dify_config.DEPLOY_ENV)
return response
@app.route("/health")
def health():
return Response(
json.dumps({"pid": os.getpid(), "status": "ok", "version": dify_config.CURRENT_VERSION}),
status=200,
content_type="application/json",
)
@app.route("/threads")
def threads():
num_threads = threading.active_count()
threads = threading.enumerate()
thread_list = []
for thread in threads:
thread_name = thread.name
thread_id = thread.ident
is_alive = thread.is_alive()
thread_list.append(
{
"name": thread_name,
"id": thread_id,
"is_alive": is_alive,
}
)
return {
"pid": os.getpid(),
"thread_num": num_threads,
"threads": thread_list,
}
@app.route("/db-pool-stat")
def pool_stat():
engine = db.engine
return {
"pid": os.getpid(),
"pool_size": engine.pool.size(),
"checked_in_connections": engine.pool.checkedin(),
"checked_out_connections": engine.pool.checkedout(),
"overflow_connections": engine.pool.overflow(),
"connection_timeout": engine.pool.timeout(),
"recycle_time": db.engine.pool._recycle,
}
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5001)

View File

@ -1,55 +1,15 @@
import logging
import os
import time
from configs import dify_config
if not dify_config.DEBUG:
from gevent import monkey
monkey.patch_all()
import grpc.experimental.gevent
grpc.experimental.gevent.init_gevent()
import json
from flask import Flask, Response, request
from flask_cors import CORS
from flask_login import user_loaded_from_request, user_logged_in
from werkzeug.exceptions import Unauthorized
import contexts
from commands import register_commands
from configs import dify_config
from extensions import (
ext_celery,
ext_code_based_extension,
ext_compress,
ext_database,
ext_hosting_provider,
ext_logging,
ext_login,
ext_mail,
ext_migrate,
ext_proxy_fix,
ext_redis,
ext_sentry,
ext_storage,
)
from extensions.ext_database import db
from extensions.ext_login import login_manager
from libs.passport import PassportService
from services.account_service import AccountService
class DifyApp(Flask):
pass
from dify_app import DifyApp
# ----------------------------
# Application Factory Function
# ----------------------------
def create_flask_app_with_configs() -> Flask:
def create_flask_app_with_configs() -> DifyApp:
"""
create a raw flask app
with configs loaded from .env file
@ -69,117 +29,72 @@ def create_flask_app_with_configs() -> Flask:
return dify_app
def create_app() -> Flask:
def create_app() -> DifyApp:
start_time = time.perf_counter()
app = create_flask_app_with_configs()
app.secret_key = dify_config.SECRET_KEY
initialize_extensions(app)
register_blueprints(app)
register_commands(app)
end_time = time.perf_counter()
if dify_config.DEBUG:
logging.info(f"Finished create_app ({round((end_time - start_time) * 1000, 2)} ms)")
return app
def initialize_extensions(app):
# Since the application instance is now created, pass it to each Flask
# extension instance to bind it to the Flask application instance (app)
ext_logging.init_app(app)
ext_compress.init_app(app)
ext_code_based_extension.init()
ext_database.init_app(app)
ext_migrate.init(app, db)
ext_redis.init_app(app)
ext_storage.init_app(app)
ext_celery.init_app(app)
ext_login.init_app(app)
ext_mail.init_app(app)
ext_hosting_provider.init_app(app)
ext_sentry.init_app(app)
ext_proxy_fix.init_app(app)
# Flask-Login configuration
@login_manager.request_loader
def load_user_from_request(request_from_flask_login):
"""Load user based on the request."""
if request.blueprint not in {"console", "inner_api"}:
return None
# Check if the user_id contains a dot, indicating the old format
auth_header = request.headers.get("Authorization", "")
if not auth_header:
auth_token = request.args.get("_token")
if not auth_token:
raise Unauthorized("Invalid Authorization token.")
else:
if " " not in auth_header:
raise Unauthorized("Invalid Authorization header format. Expected 'Bearer <api-key>' format.")
auth_scheme, auth_token = auth_header.split(None, 1)
auth_scheme = auth_scheme.lower()
if auth_scheme != "bearer":
raise Unauthorized("Invalid Authorization header format. Expected 'Bearer <api-key>' format.")
decoded = PassportService().verify(auth_token)
user_id = decoded.get("user_id")
logged_in_account = AccountService.load_logged_in_account(account_id=user_id)
return logged_in_account
@user_logged_in.connect
@user_loaded_from_request.connect
def on_user_logged_in(_sender, user):
"""Called when a user logged in."""
if user:
contexts.tenant_id.set(user.current_tenant_id)
@login_manager.unauthorized_handler
def unauthorized_handler():
"""Handle unauthorized requests."""
return Response(
json.dumps({"code": "unauthorized", "message": "Unauthorized."}),
status=401,
content_type="application/json",
def initialize_extensions(app: DifyApp):
from extensions import (
ext_app_metrics,
ext_blueprints,
ext_celery,
ext_code_based_extension,
ext_commands,
ext_compress,
ext_database,
ext_hosting_provider,
ext_import_modules,
ext_logging,
ext_login,
ext_mail,
ext_migrate,
ext_proxy_fix,
ext_redis,
ext_sentry,
ext_set_secretkey,
ext_storage,
ext_timezone,
ext_warnings,
)
extensions = [
ext_timezone,
ext_logging,
ext_warnings,
ext_import_modules,
ext_set_secretkey,
ext_compress,
ext_code_based_extension,
ext_database,
ext_app_metrics,
ext_migrate,
ext_redis,
ext_storage,
ext_celery,
ext_login,
ext_mail,
ext_hosting_provider,
ext_sentry,
ext_proxy_fix,
ext_blueprints,
ext_commands,
]
for ext in extensions:
short_name = ext.__name__.split(".")[-1]
is_enabled = ext.is_enabled() if hasattr(ext, "is_enabled") else True
if not is_enabled:
if dify_config.DEBUG:
logging.info(f"Skipped {short_name}")
continue
# register blueprint routers
def register_blueprints(app):
from controllers.console import bp as console_app_bp
from controllers.files import bp as files_bp
from controllers.inner_api import bp as inner_api_bp
from controllers.service_api import bp as service_api_bp
from controllers.web import bp as web_bp
CORS(
service_api_bp,
allow_headers=["Content-Type", "Authorization", "X-App-Code"],
methods=["GET", "PUT", "POST", "DELETE", "OPTIONS", "PATCH"],
)
app.register_blueprint(service_api_bp)
CORS(
web_bp,
resources={r"/*": {"origins": dify_config.WEB_API_CORS_ALLOW_ORIGINS}},
supports_credentials=True,
allow_headers=["Content-Type", "Authorization", "X-App-Code"],
methods=["GET", "PUT", "POST", "DELETE", "OPTIONS", "PATCH"],
expose_headers=["X-Version", "X-Env"],
)
app.register_blueprint(web_bp)
CORS(
console_app_bp,
resources={r"/*": {"origins": dify_config.CONSOLE_CORS_ALLOW_ORIGINS}},
supports_credentials=True,
allow_headers=["Content-Type", "Authorization"],
methods=["GET", "PUT", "POST", "DELETE", "OPTIONS", "PATCH"],
expose_headers=["X-Version", "X-Env"],
)
app.register_blueprint(console_app_bp)
CORS(files_bp, allow_headers=["Content-Type"], methods=["GET", "PUT", "POST", "DELETE", "OPTIONS", "PATCH"])
app.register_blueprint(files_bp)
app.register_blueprint(inner_api_bp)
start_time = time.perf_counter()
ext.init_app(app)
end_time = time.perf_counter()
if dify_config.DEBUG:
logging.info(f"Loaded {short_name} ({round((end_time - start_time) * 1000, 2)} ms)")

View File

@ -640,15 +640,3 @@ where sites.id is null limit 1000"""
break
click.echo(click.style("Fix for missing app-related sites completed successfully!", fg="green"))
def register_commands(app):
app.cli.add_command(reset_password)
app.cli.add_command(reset_email)
app.cli.add_command(reset_encrypt_key_pair)
app.cli.add_command(vdb_migrate)
app.cli.add_command(convert_to_agent_apps)
app.cli.add_command(add_qdrant_doc_id_index)
app.cli.add_command(create_tenant)
app.cli.add_command(upgrade_db)
app.cli.add_command(fix_app_site_missing)

View File

@ -17,11 +17,6 @@ class DeploymentConfig(BaseSettings):
default=False,
)
TESTING: bool = Field(
description="Enable testing mode for running automated tests",
default=False,
)
EDITION: str = Field(
description="Deployment edition of the application (e.g., 'SELF_HOSTED', 'CLOUD')",
default="SELF_HOSTED",

5
api/dify_app.py Normal file
View File

@ -0,0 +1,5 @@
from flask import Flask
class DifyApp(Flask):
pass

View File

@ -0,0 +1,65 @@
import json
import os
import threading
from flask import Response
from configs import dify_config
from dify_app import DifyApp
def init_app(app: DifyApp):
@app.after_request
def after_request(response):
"""Add Version headers to the response."""
response.headers.add("X-Version", dify_config.CURRENT_VERSION)
response.headers.add("X-Env", dify_config.DEPLOY_ENV)
return response
@app.route("/health")
def health():
return Response(
json.dumps({"pid": os.getpid(), "status": "ok", "version": dify_config.CURRENT_VERSION}),
status=200,
content_type="application/json",
)
@app.route("/threads")
def threads():
num_threads = threading.active_count()
threads = threading.enumerate()
thread_list = []
for thread in threads:
thread_name = thread.name
thread_id = thread.ident
is_alive = thread.is_alive()
thread_list.append(
{
"name": thread_name,
"id": thread_id,
"is_alive": is_alive,
}
)
return {
"pid": os.getpid(),
"thread_num": num_threads,
"threads": thread_list,
}
@app.route("/db-pool-stat")
def pool_stat():
from extensions.ext_database import db
engine = db.engine
return {
"pid": os.getpid(),
"pool_size": engine.pool.size(),
"checked_in_connections": engine.pool.checkedin(),
"checked_out_connections": engine.pool.checkedout(),
"overflow_connections": engine.pool.overflow(),
"connection_timeout": engine.pool.timeout(),
"recycle_time": db.engine.pool._recycle,
}

View File

@ -0,0 +1,48 @@
from configs import dify_config
from dify_app import DifyApp
def init_app(app: DifyApp):
# register blueprint routers
from flask_cors import CORS
from controllers.console import bp as console_app_bp
from controllers.files import bp as files_bp
from controllers.inner_api import bp as inner_api_bp
from controllers.service_api import bp as service_api_bp
from controllers.web import bp as web_bp
CORS(
service_api_bp,
allow_headers=["Content-Type", "Authorization", "X-App-Code"],
methods=["GET", "PUT", "POST", "DELETE", "OPTIONS", "PATCH"],
)
app.register_blueprint(service_api_bp)
CORS(
web_bp,
resources={r"/*": {"origins": dify_config.WEB_API_CORS_ALLOW_ORIGINS}},
supports_credentials=True,
allow_headers=["Content-Type", "Authorization", "X-App-Code"],
methods=["GET", "PUT", "POST", "DELETE", "OPTIONS", "PATCH"],
expose_headers=["X-Version", "X-Env"],
)
app.register_blueprint(web_bp)
CORS(
console_app_bp,
resources={r"/*": {"origins": dify_config.CONSOLE_CORS_ALLOW_ORIGINS}},
supports_credentials=True,
allow_headers=["Content-Type", "Authorization"],
methods=["GET", "PUT", "POST", "DELETE", "OPTIONS", "PATCH"],
expose_headers=["X-Version", "X-Env"],
)
app.register_blueprint(console_app_bp)
CORS(files_bp, allow_headers=["Content-Type"], methods=["GET", "PUT", "POST", "DELETE", "OPTIONS", "PATCH"])
app.register_blueprint(files_bp)
app.register_blueprint(inner_api_bp)

View File

@ -3,12 +3,12 @@ from datetime import timedelta
import pytz
from celery import Celery, Task
from celery.schedules import crontab
from flask import Flask
from configs import dify_config
from dify_app import DifyApp
def init_app(app: Flask) -> Celery:
def init_app(app: DifyApp) -> Celery:
class FlaskTask(Task):
def __call__(self, *args: object, **kwargs: object) -> object:
with app.app_context():

View File

@ -1,7 +1,8 @@
from core.extension.extension import Extension
from dify_app import DifyApp
def init():
def init_app(app: DifyApp):
code_based_extension.init()

View File

@ -0,0 +1,29 @@
from dify_app import DifyApp
def init_app(app: DifyApp):
from commands import (
add_qdrant_doc_id_index,
convert_to_agent_apps,
create_tenant,
fix_app_site_missing,
reset_email,
reset_encrypt_key_pair,
reset_password,
upgrade_db,
vdb_migrate,
)
cmds_to_register = [
reset_password,
reset_email,
reset_encrypt_key_pair,
vdb_migrate,
convert_to_agent_apps,
add_qdrant_doc_id_index,
create_tenant,
upgrade_db,
fix_app_site_missing,
]
for cmd in cmds_to_register:
app.cli.add_command(cmd)

View File

@ -1,10 +1,12 @@
from flask import Flask
from configs import dify_config
from dify_app import DifyApp
def init_app(app: Flask):
if dify_config.API_COMPRESSION_ENABLED:
def is_enabled() -> bool:
return dify_config.API_COMPRESSION_ENABLED
def init_app(app: DifyApp):
from flask_compress import Compress
compress = Compress()

View File

@ -1,6 +1,8 @@
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import MetaData
from dify_app import DifyApp
POSTGRES_INDEXES_NAMING_CONVENTION = {
"ix": "%(column_0_label)s_idx",
"uq": "%(table_name)s_%(column_0_name)s_key",
@ -13,5 +15,5 @@ metadata = MetaData(naming_convention=POSTGRES_INDEXES_NAMING_CONVENTION)
db = SQLAlchemy(metadata=metadata)
def init_app(app):
def init_app(app: DifyApp):
db.init_app(app)

View File

@ -1,9 +1,10 @@
from flask import Flask
from core.hosting_configuration import HostingConfiguration
hosting_configuration = HostingConfiguration()
def init_app(app: Flask):
from dify_app import DifyApp
def init_app(app: DifyApp):
hosting_configuration.init_app(app)

View File

@ -0,0 +1,6 @@
from dify_app import DifyApp
def init_app(app: DifyApp):
from events import event_handlers # noqa: F401
from models import account, dataset, model, source, task, tool, tools, web # noqa: F401

View File

@ -3,12 +3,11 @@ import os
import sys
from logging.handlers import RotatingFileHandler
from flask import Flask
from configs import dify_config
from dify_app import DifyApp
def init_app(app: Flask):
def init_app(app: DifyApp):
log_handlers = []
log_file = dify_config.LOG_FILE
if log_file:

View File

@ -1,7 +1,62 @@
import json
import flask_login
from flask import Response, request
from flask_login import user_loaded_from_request, user_logged_in
from werkzeug.exceptions import Unauthorized
import contexts
from dify_app import DifyApp
from libs.passport import PassportService
from services.account_service import AccountService
login_manager = flask_login.LoginManager()
def init_app(app):
# Flask-Login configuration
@login_manager.request_loader
def load_user_from_request(request_from_flask_login):
"""Load user based on the request."""
if request.blueprint not in {"console", "inner_api"}:
return None
# Check if the user_id contains a dot, indicating the old format
auth_header = request.headers.get("Authorization", "")
if not auth_header:
auth_token = request.args.get("_token")
if not auth_token:
raise Unauthorized("Invalid Authorization token.")
else:
if " " not in auth_header:
raise Unauthorized("Invalid Authorization header format. Expected 'Bearer <api-key>' format.")
auth_scheme, auth_token = auth_header.split(None, 1)
auth_scheme = auth_scheme.lower()
if auth_scheme != "bearer":
raise Unauthorized("Invalid Authorization header format. Expected 'Bearer <api-key>' format.")
decoded = PassportService().verify(auth_token)
user_id = decoded.get("user_id")
logged_in_account = AccountService.load_logged_in_account(account_id=user_id)
return logged_in_account
@user_logged_in.connect
@user_loaded_from_request.connect
def on_user_logged_in(_sender, user):
"""Called when a user logged in."""
if user:
contexts.tenant_id.set(user.current_tenant_id)
@login_manager.unauthorized_handler
def unauthorized_handler():
"""Handle unauthorized requests."""
return Response(
json.dumps({"code": "unauthorized", "message": "Unauthorized."}),
status=401,
content_type="application/json",
)
def init_app(app: DifyApp):
login_manager.init_app(app)

View File

@ -1,10 +1,10 @@
import logging
from typing import Optional
import resend
from flask import Flask
from configs import dify_config
from dify_app import DifyApp
class Mail:
@ -26,6 +26,8 @@ class Mail:
match mail_type:
case "resend":
import resend
api_key = dify_config.RESEND_API_KEY
if not api_key:
raise ValueError("RESEND_API_KEY is not set")
@ -84,7 +86,11 @@ class Mail:
)
def init_app(app: Flask):
def is_enabled() -> bool:
return dify_config.MAIL_TYPE is not None and dify_config.MAIL_TYPE != ""
def init_app(app: DifyApp):
mail.init_app(app)

View File

@ -1,5 +1,9 @@
import flask_migrate
from dify_app import DifyApp
def init(app, db):
def init_app(app: DifyApp):
import flask_migrate
from extensions.ext_database import db
flask_migrate.Migrate(app, db)

View File

@ -1,9 +1,8 @@
from flask import Flask
from configs import dify_config
from dify_app import DifyApp
def init_app(app: Flask):
def init_app(app: DifyApp):
if dify_config.RESPECT_XFORWARD_HEADERS_ENABLED:
from werkzeug.middleware.proxy_fix import ProxyFix

View File

@ -4,6 +4,7 @@ from redis.connection import Connection, SSLConnection
from redis.sentinel import Sentinel
from configs import dify_config
from dify_app import DifyApp
class RedisClientWrapper:
@ -43,7 +44,7 @@ class RedisClientWrapper:
redis_client = RedisClientWrapper()
def init_app(app):
def init_app(app: DifyApp):
global redis_client
connection_class = Connection
if dify_config.REDIS_USE_SSL:

View File

@ -1,15 +1,19 @@
import openai
import sentry_sdk
from langfuse import parse_error
from sentry_sdk.integrations.celery import CeleryIntegration
from sentry_sdk.integrations.flask import FlaskIntegration
from werkzeug.exceptions import HTTPException
from configs import dify_config
from core.model_runtime.errors.invoke import InvokeRateLimitError
from dify_app import DifyApp
def before_send(event, hint):
def init_app(app: DifyApp):
if dify_config.SENTRY_DSN:
import openai
import sentry_sdk
from langfuse import parse_error
from sentry_sdk.integrations.celery import CeleryIntegration
from sentry_sdk.integrations.flask import FlaskIntegration
from werkzeug.exceptions import HTTPException
from core.model_runtime.errors.invoke import InvokeRateLimitError
def before_send(event, hint):
if "exc_info" in hint:
exc_type, exc_value, tb = hint["exc_info"]
if parse_error.defaultErrorResponse in str(exc_value):
@ -17,9 +21,6 @@ def before_send(event, hint):
return event
def init_app(app):
if dify_config.SENTRY_DSN:
sentry_sdk.init(
dsn=dify_config.SENTRY_DSN,
integrations=[FlaskIntegration(), CeleryIntegration()],

View File

@ -0,0 +1,6 @@
from configs import dify_config
from dify_app import DifyApp
def init_app(app: DifyApp):
app.secret_key = dify_config.SECRET_KEY

View File

@ -5,6 +5,7 @@ from typing import Union
from flask import Flask
from configs import dify_config
from dify_app import DifyApp
from extensions.storage.base_storage import BaseStorage
from extensions.storage.storage_type import StorageType
@ -122,5 +123,5 @@ class Storage:
storage = Storage()
def init_app(app: Flask):
def init_app(app: DifyApp):
storage.init_app(app)

View File

@ -0,0 +1,11 @@
import os
import time
from dify_app import DifyApp
def init_app(app: DifyApp):
os.environ["TZ"] = "UTC"
# windows platform not support tzset
if hasattr(time, "tzset"):
time.tzset()

View File

@ -0,0 +1,7 @@
from dify_app import DifyApp
def init_app(app: DifyApp):
import warnings
warnings.simplefilter("ignore", ResourceWarning)

View File

@ -0,0 +1,19 @@
from configs import dify_config
def apply_gevent_threading_patch():
"""
Run threading patch by gevent
to make standard library threading compatible.
Patching should be done as early as possible in the lifecycle of the program.
:return:
"""
if not dify_config.DEBUG:
from gevent import monkey
from grpc.experimental import gevent as grpc_gevent
# gevent
monkey.patch_all()
# grpc gevent
grpc_gevent.init_gevent()

12
api/libs/version_utils.py Normal file
View File

@ -0,0 +1,12 @@
import sys
def check_supported_python_version():
python_version = sys.version_info
if not ((3, 11) <= python_version < (3, 13)):
print(
"Aborted to launch the service "
f" with unsupported Python version {python_version.major}.{python_version.minor}."
" Please ensure Python 3.11 or 3.12."
)
raise SystemExit(1)

View File

@ -71,7 +71,6 @@ def test_flask_configs(example_env_file):
assert config["EDITION"] == "SELF_HOSTED"
assert config["API_COMPRESSION_ENABLED"] is False
assert config["SENTRY_TRACES_SAMPLE_RATE"] == 1.0
assert config["TESTING"] == False
# value from env file
assert config["CONSOLE_API_URL"] == "https://example.com"

View File

@ -10,7 +10,6 @@ ABS_PATH = os.path.dirname(os.path.abspath(__file__))
PROJECT_DIR = os.path.abspath(os.path.join(ABS_PATH, os.pardir, os.pardir))
CACHED_APP = Flask(__name__)
CACHED_APP.config.update({"TESTING": True})
@pytest.fixture