Update project

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 import Celery
+from celery.schedules import crontab
+
+
+def init_celery(app: Flask) -> None:
+    """
+    Initialize Celery for the Flask application.
+
+    :param app: The Flask application.
+    :type app: Flask
+    """
+    celery = Celery(
+        app.import_name,
+        backend=app.config['CELERY_RESULT_BACKEND'],
+        broker=app.config['CELERY_BROKER_URL']
+    )
+    celery.conf.update(app.config)
+
+    class ContextTask(celery.Task):
+        def __call__(self, *args, **kwargs):
+            with app.app_context():
+                return self.run(*args, **kwargs)
+
+    celery.Task = ContextTask
+
+    @celery.task()
+    def update_database():
+        # Add your database operations here
+        pass
+
+    celery.conf.beat_schedule = {
+        'update-database-every-day': {
+            'task': 'update_database',
+            'schedule': crontab(hour=0, minute=0)  # Execute daily at midnight
+        }
+    }
+
+    app.celery = celery

application/extensions/init_sqlalchemy.py

@@ -13,7 +13,7 @@
 from flask import Flask
 from flask_sqlalchemy import SQLAlchemy
 
-from application.utils import DatabaseURI
+from application.utils import DatabaseUri
 
 db = SQLAlchemy()
 
@@ -26,7 +26,7 @@ def init_database(app: Flask) -> None:
     :return: None
     """
     cfg = app.config.Database
-    uri = DatabaseURI(
+    uri = DatabaseUri(
         db_type=cfg.Type,
         username=cfg.Username,
         password=cfg.Password,

application/utils/__init__.py

@@ -10,5 +10,5 @@
 # @Description :
 """
 
-from .dsn import DatabaseURI
+from .dsn import DatabaseUri
 from .elasticsearch import ElasticsearchUtil

application/utils/dsn/__init__.py

@@ -10,4 +10,5 @@
 # @Description :
 """
 
-from .dsn import DatabaseURI
+from .redis import RedisUri
+from .sql import DatabaseUri

application/utils/dsn/redis.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 15:47
+# @File        : redis.py
+# @Description :
+"""
+
+
+class RedisUri:
+    """
+    A class to generate a Redis connection URI.
+    """
+
+    def __init__(self, host='localhost', port=6379, db=0, username=None, password=None):
+        """
+        Initializes a new instance of the RedisUri class.
+
+        :param host: The hostname of the Redis server. Default is 'localhost'.
+        :type host: str
+        :param port: The port number to connect to the Redis server. Default is 6379.
+        :type port: int
+        :param db: The database number to connect to on the Redis server. Default is 0.
+        :type db: int
+        :param username: The username for authentication to the Redis server. Default is None.
+        :type username: str
+        :param password: The password for authentication to the Redis server. Default is None.
+        :type password: str
+        """
+        self.host = host
+        self.port = port
+        self.db = db
+        self.username = username
+        self.password = password
+
+    def create(self) -> str:
+        """
+        Creates and returns the Redis connection URI.
+
+        :return: The Redis connection URI.
+        :rtype: str
+        """
+        if self.username and self.password:
+            return f'redis://{self.username}:{self.password}@{self.host}:{self.port}/{self.db}'
+        else:
+            return f'redis://{self.host}:{self.port}/{self.db}'
+
+    def __repr__(self):
+        """
+        Returns a formal string representation of the RedisUri object.
+
+        :return: The formal string representation of the RedisUri object.
+        :rtype: str
+        """
+        return (f'RedisUri(host={self.host}, port={self.port}, db={self.db}, username={self.username}, '
+                f'password={self.password})')
+
+    def __str__(self):
+        """
+        Returns a string representation of the RedisUri object.
+
+        :return: The string representation of the RedisUri object.
+        :rtype: str
+        """
+        return self.create()

application/utils/dsn/dsn.py → application/utils/dsn/sql.py

@@ -6,12 +6,12 @@
 # @Author      : StudentCWZ
 # @Email       : StudentCWZ@outlook.com
 # @Date        : 2023/10/29 00:36
-# @File        : dsn.py
+# @File        : sql.py
 # @Description :
 """
 
 
-class DatabaseURI:
+class DatabaseUri:
     """
     The DatabaseURI class is a utility for generating SQLAlchemy database
     connection strings for various types of databases.
@@ -60,7 +60,7 @@ class DatabaseURI:
                 return f'{self.db_type}://{self.username}:{self.password}@{self.host}:{self.port}/{self.db}'
 
     def __repr__(self):
-        return f'<DatabaseURI({self.db_type})>'
+        return f'<DatabaseUri({self.db_type})>'
 
     def __str__(self):
         return self.create()