'报表充值列表'

app/api/statement/crud.py

+from typing import Optional
 from sqlalchemy.orm import Session
+
+from core.config.env import env
+from libs.db_link import LinkMysql
 from models import statement as models
 from app.api.statement import schemas
+from sqlalchemy import and_
 
 
 def get_statement(db: Session, user_id: int):
@@ -11,15 +16,12 @@ def get_statement_by_name(db: Session, username: str):
     return db.query(models.Statement).filter(models.Statement.username == username).first()
 
 
-def get_statements(db: Session, page: int = 0, size: int = 10):
-    return db.query(models.Statement).offset(page).limit(size).all()
-
+def get_statements(param):
+    page = param.page if param.page else 1
+    size = param.size if param.size else 10
+    sql = f"select order_number,userid,user_number,nick_name,money,paychannel,sid,lastupdate,current,payment_time " \
+          f"from orders o INNER JOIN v2_user v on o.userid = v.user_id where o.order_number LIKE '%{param.get('order_number')}%' and " \
+          f"o.userid LIKE '%{param.get('user_id')}%' and o.sid LIKE '%{param.get('sid')}%' and o.paychannel LIKE '%{param.get('paychannel')}%' and orcurrent>='{param.get('start_time')}' and current <= '{param.get('end_time')}' LIMIT {(int(page)-1) * size},{page * size}"
 
-def create_statementr(db: Session, user: schemas.StatementCreate):
-    db_user = models.Statement(order_number=user.order_number, user_id=user.user_id, zb_number=user.zb_number,
-                               nickname=user.nickname, recharge_amount=user.recharge_amount, recharge_channel=user.recharge_channel,
-                               flowing_water_number=user.flowing_water_number)
-    db.add(db_user)
-    db.commit()
-    db.refresh(db_user)
-    return db_user
+    query_res = LinkMysql(env.MysqlDB).query_mysql(sql)
+    return query_res

app/api/statement/schemas.py

+from typing import Optional
+
 from pydantic import BaseModel
 
 
@@ -15,6 +17,17 @@ class StatementCreate(StatementBase):
     flowing_water_number: int
 
 
+class StatementList(BaseModel):
+    page: Optional[int] = None
+    size: Optional[int] = None
+    start_time:  Optional[str] = ""
+    end_time: Optional[str] = ""
+    order_number: Optional[str] = ""
+    user_id: Optional[int] = None
+    flowing_water_number: Optional[str] = ""
+    recharge_channel: Optional[str] = ""
+
+
 class Statement(StatementBase):
     id: int
 

app/api/statement/views.py

@@ -2,18 +2,17 @@ from app import get_db
 
 from sqlalchemy.orm import Session
 from app.api.statement import crud, schemas
-from fastapi import Depends, APIRouter, HTTPException
-from libs.result_format import HttpResultResponse, HttpMessage
-from typing import Optional
+from fastapi import Depends, APIRouter, Request
+from libs.result_format import HttpResultResponse
+
 router = APIRouter()
 
 
 @router.post("/recharge/list")
-def statement_recharge_list(page: int = 1, size: int = 10, db: Session = Depends(get_db)):
+def statement_recharge_list(data: schemas.StatementList):
     """充值报表列表"""
-
-    users = crud.get_statements(db, page=page, size=size)
-    return users
+    statement_list = crud.get_statements(data)
+    return HttpResultResponse(data=statement_list)
 
 
 @router.post("/create")

core/config/env.py

@@ -13,6 +13,13 @@ class Env(BaseSettings):
     DATABASE_URI: str = 'sqlite://:memory:'
     DATABASE_USER: str = ''
     DATABASE_PWD: str = ''
+    MysqlDB: dict = {
+        "database": "3yakj_v2",
+        "host": "106.55.103.148",
+        "port": 3398,
+        "pwd": "c1ea602311a369f6",
+        "user": "root"
+    }
 
 
 class TestingEnv(Env):

libs/db_link.py

+import pymysql
+from DBUtils.PooledDB import PooledDB
+
+
+# 连接mysql
+class LinkMysql(object):
+    def __init__(self, db_info):
+        self.POOLMYSQL = PooledDB(
+            creator=pymysql,
+            maxconnections=6,
+            mincached=2,
+            blocking=True,
+            ping=0,
+
+            host=db_info.get("host"),
+            port=db_info.get("port"),
+            user=db_info.get("user"),
+            password=db_info.get("pwd"),
+            database=db_info.get("database"),
+            charset='utf8'
+        )
+
+    # 查询
+    def query_mysql(self, sql):
+        conn = self.POOLMYSQL.connection()
+        cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
+        cursor.execute(sql)
+        result = cursor.fetchall()
+        cursor.close()
+        conn.close()
+        return result
+
+    # 修改
+    def perform_mysql(self, sql):
+        conn = self.POOLMYSQL.connection()
+        cursor = conn.cursor()
+        cursor.execute(sql)
+        last_id = cursor.lastrowid
+        conn.commit()
+        cursor.close()
+        conn.close()
+        return last_id