Update project

application/__init__.py

@@ -18,7 +18,7 @@ from application.extensions import init_plugs
 from application.views import init_views
 from application.script import init_script
 from application.utils import make_celery
-from celery.schedules import crontab
+from application.common import config
 
 
 def create_app() -> Flask:
@@ -38,22 +38,8 @@ def create_app() -> Flask:
     # This could include things like database connectors, authentication systems, etc.
     init_plugs(app)
 
-    celery = make_celery(app)
-    celery.conf.update(app.config)
-
-    @celery.task(bind=True)
-    def update_database(self):
-        # Add your database operations here
-        print("update database")
-
-    celery.conf.beat_schedule = {
-        'update-database-every-minute': {
-            'task': 'application.__init__.update_database',
-            'schedule': crontab(minute='*')  # Execute every minute
-        }
-    }
-
-    app.celery = celery
+    # Set universal config
+    config.Universal = app.config
 
     # Register the routes that this application will respond to.
     # This includes both the route URLs and the handlers for each route.

application/common/__init__.py

@@ -10,6 +10,4 @@
 # @Description :
 """
 
-from .config import ConfigHelper
-from .env import EnvVarHelper
-from .file import FileHelper
+from .config import Config

application/common/config.py

@@ -5,46 +5,18 @@
 # @Software    : Sublime Text 4
 # @Author      : StudentCWZ
 # @Email       : StudentCWZ@outlook.com
-# @Date        : 2023/11/3 19:43
+# @Date        : 2023/11/11 14:11
 # @File        : config.py
 # @Description :
 """
 
 
-from typing import Any
+# config.py
+class Config:
+    """Configuration for the application."""
+    def __init__(self):
+        self.Universal = None
 
-from flask import Flask
-from loguru import logger
 
-from application.libs import ConfigKeyError
-
-
-class ConfigHelper:
-    """
-    The ConfigHelper class is a utility for fetching configuration values
-    from a Flask application.
-
-    :param app: The Flask application instance from which to fetch configuration values.
-    """
-
-    def __init__(self, app: Flask):
-        self.app = app
-
-    def __getattr__(self, name: str) -> Any:
-        """
-        Get a config value as an attribute.
-
-        :param name: The name of the config key.
-        :return: The value for the provided key, or None if it does not exist.
-        """
-        try:
-            return self.app.config[name]
-        except KeyError:
-            logger.error(f'Key {name} not found in configuration')
-            raise ConfigKeyError(name)
-
-    def __repr__(self):
-        return f"<ConfigHelper with app {self.app}>"
-
-    def __str__(self):
-        return f"ConfigHelper for app {self.app}"
+# This creates a global instance of the Config class
+config = Config()

application/extensions/__init__.py

@@ -21,7 +21,6 @@ from .init_migrate import init_migrate
 from .init_apispec import init_apispec
 from .init_marshmallow import init_marshmallow
 from .init_elasticsearch import init_elasticsearch
-# from .init_celery import init_celery
 
 
 def init_plugs(app: Flask) -> None:
@@ -34,4 +33,3 @@ def init_plugs(app: Flask) -> None:
     init_marshmallow(app)
     init_elasticsearch(app)
     init_cors(app)
-    # init_celery(app)

application/extensions/init_celery.py

-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-"""
-# @Version     : Python 3.11.4
-# @Software    : Sublime Text 4
-# @Author      : StudentCWZ
-# @Email       : StudentCWZ@outlook.com
-# @Date        : 2023/11/7 16:10
-# @File        : init_celery.py
-# @Description :
-"""
-
-from flask import Flask
-from celery.schedules import crontab
-
-from application.utils import make_celery
-
-
-def init_celery(app: Flask) -> None:
-    celery = make_celery(app)
-    app.celery = celery
-
-    celery.conf.beat_schedule = {
-        'update-database-every-day': {
-            'task': 'application.libs.tasks.task.update_database',
-            'schedule': crontab(minute='*')  # Execute every minute
-        }
-    }

application/extensions/init_sqlalchemy.py

@@ -36,4 +36,7 @@ def init_database(app: Flask) -> None:
         driver=cfg.Driver,
     )
     app.config.setdefault('SQLALCHEMY_DATABASE_URI', uri.create())
+    db.app = app
     db.init_app(app)
+    with app.app_context():
+        db.create_all()

application/libs/flask_elasticsearch/elasticsearch.py

@@ -19,7 +19,7 @@ from flask import _app_ctx_stack as stack_context
 from flask_elasticsearch import FlaskElasticsearch as BaseFlaskElasticsearch
 from loguru import logger
 
-from application.common import ConfigHelper
+from application.libs.helper import ConfigHelper
 
 
 class FlaskElasticsearch(BaseFlaskElasticsearch):

application/libs/flask_loguru/logger.py

@@ -17,7 +17,7 @@ from flask import Flask, request, g
 from loguru import logger
 
 from .format import patching
-from application.common import ConfigHelper
+from application.libs.helper import ConfigHelper
 
 
 class FlaskLoguru:

application/libs/helper/__init__.py

+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+# @Version     : Python 3.11.4
+# @Software    : Sublime Text 4
+# @Author      : StudentCWZ
+# @Email       : StudentCWZ@outlook.com
+# @Date        : 2023/11/11 14:05
+# @File        : __init__.py
+# @Description :
+"""
+
+from .config import ConfigHelper
+from .env import EnvVarHelper
+from .file import FileHelper

application/libs/helper/config.py

+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+# @Version     : Python 3.11.4
+# @Software    : Sublime Text 4
+# @Author      : StudentCWZ
+# @Email       : StudentCWZ@outlook.com
+# @Date        : 2023/11/3 19:43
+# @File        : config.py
+# @Description :
+"""
+
+
+from typing import Any
+
+from flask import Flask
+from loguru import logger
+
+from application.libs import ConfigKeyError
+
+
+class ConfigHelper:
+    """
+    The ConfigHelper class is a utility for fetching configuration values
+    from a Flask application.
+
+    :param app: The Flask application instance from which to fetch configuration values.
+    """
+
+    def __init__(self, app: Flask):
+        self.app = app
+
+    def __getattr__(self, name: str) -> Any:
+        """
+        Get a config value as an attribute.
+
+        :param name: The name of the config key.
+        :return: The value for the provided key, or None if it does not exist.
+        """
+        try:
+            return self.app.config[name]
+        except KeyError:
+            logger.error(f'Key {name} not found in configuration')
+            raise ConfigKeyError(name)
+
+    def __repr__(self):
+        return f"<ConfigHelper with app {self.app}>"
+
+    def __str__(self):
+        return f"ConfigHelper for app {self.app}"

application/libs/tasks/task.py

@@ -9,13 +9,3 @@
 # @File        : task.py
 # @Description :
 """
-
-
-from flask import current_app
-
-
-@current_app.celery.task
-def update_database():
-    # Add your database operations here
-    print(123312312)
-    pass

application/utils/loaders/consul_loader.py

@@ -16,8 +16,8 @@ import requests
 from dynaconf.base import LazySettings
 from dynaconf.utils.parse_conf import parse_conf_data
 
-from application.common import EnvVarHelper
 from application.libs import ConsulConfig
+from application.libs.helper import EnvVarHelper
 
 IDENTIFIER = "consul_loader"
 

application/utils/loaders/yaml_loader.py

@@ -17,8 +17,8 @@ from typing import Union
 from dynaconf.base import LazySettings
 from dynaconf.utils.parse_conf import parse_conf_data
 
-from application.common import FileHelper
 from application.libs import LocalConfig
+from application.libs.helper import FileHelper
 
 IDENTIFIER = "yaml_loader"
 

application/views/user/user.py

@@ -18,7 +18,7 @@ from marshmallow import ValidationError
 
 from application.schemas import UserSchema
 from application.utils import ElasticsearchUtil
-from application.common import ConfigHelper
+from application.libs.helper import ConfigHelper
 
 user_api = Blueprint('user_api', __name__)
 api_users = Api(user_api)
@@ -54,16 +54,16 @@ def create_user():
 
 @user_api.route('/tests', methods=['POST'])
 def test_user():
-    data = request.get_json()
-    start = data.get('start')
-    end = data.get('end')
-    dsl = ElasticsearchUtil.dsl(start, end)
-    index = current_app.config.Elasticsearch.Index
-    res = ElasticsearchUtil.search(index, dsl)
-    print(len(res))
-    data = {
+    # data = request.get_json()
+    # start = data.get('start')
+    # end = data.get('end')
+    # dsl = ElasticsearchUtil.dsl(start, end)
+    # index = current_app.config.Elasticsearch.Index
+    # res = ElasticsearchUtil.search(index, dsl)
+    # print(len(res))
+    res = {
         "user_name": "libai",
         "user_age": 18,
     }
-    res_json = json.dumps(data)
+    res_json = json.dumps(res)
     return res_json, 200