main.py

@@ -12,7 +12,7 @@ from starlette.middleware.cors import CORSMiddleware
 
 from LocalModel import CustomLogin, SaveUser, QueryUser, DeleteUser
 from logic import *
-from model import CustomUser, UserInfo, database
+from model import CustomUser, UserInfo, ZaneTest, database
 
 API_KEY = "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"
 
@@ -296,9 +296,12 @@ async def test(request: Request):
     request_origin = request.headers.get('origin')
     if request_origin is None:
         request_origin = "unknown"
-    content = {"message": "Hello World" + request_origin, "db": "disconnect！！！"}
+    content = {"message": "Hello World" +
+               request_origin, "db": "disconnect！！！"}
     headers = {'Access-Control-Allow-Origin': request_origin}
-    content["db"] = "is_closed: " + str(database.is_closed()) + " is_usable:" + str(database.is_connection_usable())
+    content["db"] = "is_closed: " + \
+        str(database.is_closed()) + " is_usable:" + \
+        str(database.is_connection_usable())
     try:
         dt = CustomUser.select()
         for item in dt:
@@ -310,6 +313,6 @@ async def test(request: Request):
 
 @app.get("/api/test2")
 async def test2():
-    users = UserInfo.select().where(UserInfo.customer != None)
+    users = ZaneTest.select().where(ZaneTest.a != None)
     users = [__build_user_object(dt) for dt in users]
     return {"message": "Hello World", "users": users}

model.py

@@ -1,13 +1,13 @@
 from peewee import *
 from playhouse.pool import PooledMySQLDatabase
 
-database = PooledMySQLDatabase('yixue',
-                               user='yixue',
+database = PooledMySQLDatabase('yixue_test',
+                               user='yixue_test',
                                password='Lai123',
                                host='test.db.cxhy.cn',
-                               port=3306,
+                               port=3307,
                                max_connections=32,
-                               stale_timeout=300, )
+                               stale_timeout=5, )
 
 
 class UnknownField(object):
@@ -262,6 +262,13 @@ class Tianyiguiren(BaseModel):
         table_name = 'tianyiguiren'
 
 
+class ZaneTest(BaseModel):
+    a = IntegerField(null=True)
+
+    class Meta:
+        table_name = "zane_test"
+
+
 class UserInfo(BaseModel):
     beizhu = TextField(null=True)
     customer = TextField(null=True)

test.py

@@ -0,0 +1,31 @@
+
+import time
+import requests
+
+import threading
+
+
+def http_get(url):
+    response = requests.get(url)
+    resp = response.text
+    print(f"Response: {len(resp)}")
+
+
+def test():
+    print('This is a test function in test.py')
+
+    url = "https://yixue.cxhy.cn/api/test2"
+
+    while True:
+        for i in range(100):
+            t = threading.Thread(target=http_get, args=(url,))
+            t.setDaemon(True)
+            t.start()
+            print(f"Thread {i} started")
+            # r = http_get(url)
+            # print(len(r))
+        time.sleep(3)
+
+
+if __name__ == '__main__':
+    test()