update:完善数据库和Kafka相关结构

.gitignore

@@ -1,3 +1,6 @@
+test.py
+
+
 # ---> Python
 # Byte-compiled / optimized / DLL files
 __pycache__/

conf/config.ini

@@ -1,4 +0,0 @@
-[server]
-listen = 0.0.0.0
-port = 8080
-debug = true

conf/util.py

@@ -1,8 +1,10 @@
 import configparser
 
-config_file_path = '/conf/config.ini'
+config_file_path = "config.ini"
-config = configparser.ConfigParser()
 
 
-def get_config_object(): return config
+def get_config_object():
-""" 返回配置对象 """
+    """ 返回配置对象 """
+    config = configparser.ConfigParser()
+    config.read(config_file_path, encoding='utf-8')
+    return config

config.ini

@@ -0,0 +1,16 @@
+[server]
+listen = 0.0.0.0
+port = 8080
+debug = true
+
+[database]
+host = 172.16.5.2
+port = 3306
+database = test
+user = root
+password = 123456
+
+[kafka]
+bootstrap_servers = 172.16.5.2:9092
+group_id = test
+raw_topic = raw

dao/db/example.py

@@ -0,0 +1,23 @@
+from dao.db.util import get_connet
+
+def exmaple_query():
+    with get_connet() as conn:
+        cursor = conn.cursor()
+        cursor.execute("select * from user")
+        for row in cursor.fetchall():
+            print(row)
+        cursor.close()
+
+def exmaple_insert():
+    with get_connet() as conn:
+        with conn.cursor() as cursor:
+            cursor.execute("insert into user(name, age) values('test', 18)")
+            conn.commit()
+            cursor.close()
+
+def exmaple_update():
+    with get_connet() as conn:
+        with conn.cursor() as cursor:
+            cursor.execute("update user set age=19 where name='test'")
+            conn.commit()
+            cursor.close()

dao/db/util.py

@@ -0,0 +1,15 @@
+from conf.util import get_config_object
+import pymysql
+
+conf = get_config_object()
+
+
+
+host = conf.get('database', 'host')
+port = conf.getint('database', 'port')
+user = conf.get('database', 'user')
+password = conf.get('database', 'password')
+database = conf.get('database', 'database')
+
+def get_connet() -> pymysql.Connection:
+    return pymysql.connect(host=host, port=port, user=user, password=password, database=database, charset='utf8')

dao/kafka/util.py

@@ -0,0 +1,13 @@
+from conf.util import get_config_object
+from kafka import KafkaConsumer
+
+conf = get_config_object()
+
+# Kafka
+def get_KafkaConsumer() -> KafkaConsumer:
+    """ 返回KafkaConsumer对象 """
+    consumer = KafkaConsumer(
+        bootstrap_servers=conf.get("kafka", "bootstrap_servers"),
+        group_id=conf.get("kafka", "group_id")
+    )
+    return consumer

requirements.txt

@@ -1,2 +1,3 @@
 Flask == 3.1.0
-pymysql == 1.1.1
+pymysql == 1.1.1
+confluent-kafka == 2.7.0