刚才发现每次调用完接口都有一堆连接没关掉( 如下图 ):
于是写了一段本地代码无限循环跑发现单独执行一次是可以立刻关闭连接的即便循环跑也一直是 10-30
来回跳我估计就是每次循环好就关闭了问题不大( 代码如下 ):
from sqlalchemy import create_engine, text, select
from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Session
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.pool import NullPool
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "test"
id: Mapped[int] = mapped_column(primary_key=True)
app_name: Mapped[str] = mapped_column(String(30))
while True:
# 创建数据库引擎
engine = create_engine('mysql+pymysql://root:admin4ph@192.168.3.232:33306/test', echo=True, poolclass=NullPool, isolation_level="AUTOCOMMIT")
session = Session(engine)
stmt = select(User)
for user in session.scalars(stmt):
print(user.app_name)
但是我的 flask
应用中每次请求完连接都在一直没搞懂什么问题( 代码如下 ):
from flask import current_app
from lib.database import Database
from core.database.base import Base
from lib.type import Type
from flask import g
class Platform(Database.Easy):
@property
def database(self):
"""
@ 数据库[缓存]
"""
return Database.container(**self.database_conf)["session"]
@property
def database_conf(self):
"""
@ [平台-数据库]:配置
"""
return Base.setting(current_app.config["Database"])
from sqlalchemy import create_engine, URL, delete, update, select, exists, inspect, MetaData, Table, func, text
from sqlalchemy.orm import sessionmaker, scoped_session, Session
from sqlalchemy.dialects import mysql
from core.database.base import Base
from lib.type import Type
from typing import Any
from sqlalchemy.pool import NullPool
import importlib
import re
class Database:
@classmethod
def container(cls, **config):
"""
@ 创建容器:节约开销
"""
engine = cls.create_engine(**config)
session = Session(engine)
# return Type.Database(engine=engine, session=session)
return {"engine": engine, "session": session}
@classmethod
def create_engine(cls, **kwargs):
"""
@ 创建连接:引擎
"""
return create_engine(URL.create("mysql+pymysql", **kwargs), poolclass=NullPool, isolation_level="AUTOCOMMIT")
class Easy:
def table_data_query_one(self, model: Any, condition: list = None,design:bool=False) -> dict:
"""
@ 查询:单条
:param model: 模型
:param condition: 条件
"""
condition.append(model.status >= -1 if design else model.status == 1)
result = self.database.execute(select(model).filter(*condition).limit(1)).scalar_one_or_none()
return None if result is None else result.dict()
请求指点分析什么问题