【Python系列】SQLAlchemy判断非空

💝💝💝欢迎来到我的博客，很高兴能够在这里和您见面！希望您在这里可以感受到一份轻松愉快的氛围，不仅可以获得有趣的内容和知识，也可以畅所欲言、分享您的想法和见解。

• 推荐:kwan 的首页,持续学习,不断总结,共同进步,活到老学到老
• 导航 - 檀越剑指大厂系列:全面总结 java 核心技术,jvm,并发编程 redis,kafka,Spring,微服务等- 常用开发工具系列:常用的开发工具,IDEA,Mac,Alfred,Git,typora 等- 数据库系列:详细总结了常用数据库 mysql 技术点,以及工作中遇到的 mysql 问题等- 新空间代码工作室:提供各种软件服务,承接各种毕业设计,毕业论文等- 懒人运维系列:总结好用的命令,解放双手不香吗?能用一个命令完成绝不用两个操作- 数据结构与算法系列:总结数据结构和算法,不同类型针对性训练,提升编程思维,剑指大厂

非常期待和您一起在这个小小的网络世界里共同探索、学习和成长。💝💝💝 ✨✨ 欢迎订阅本专栏 ✨✨

博客目录

sqlalchemy 查询字段非空

@staticmethod
def get_all_run_id():
    db = get_sqlalchemy_db
    try:
        with (Session(db.engine) as session):
            stmt = select(AlchemyTaskManage).where(AlchemyTaskManage.es_index_prefix.isnot(None))
            list_res = session.scalars(stmt).all()
            return [task_info.es_index_prefix for task_info in list_res] if list_res else []
    except Exception as e:
        logger.error(f"find task by name An error occurred: {e}")
        return None
    finally:
        db.close_session(session)

在数据库操作中，确保数据的完整性和准确性是非常重要的。在某些情况下，我们需要查询数据库中某个字段非空的记录。这在处理数据清洗、数据验证或者在进行数据分析时尤为常见。SQLAlchemy 是 Python 中一个非常流行的 SQL 工具包和对象关系映射（ORM）系统，它提供了一种高层的 ORM 以及底层的 SQL 表达式语言，用于数据库的交互。

在上述代码片段中，我们看到了使用 SQLAlchemy 进行数据库查询的一个例子。这段代码定义了一个名为

get_all_run_id

的静态方法，其目的是查询数据库中

es_index_prefix

字段非空的所有记录，并返回这些记录的

es_index_prefix

字段值的列表。

首先，我们来分析这段代码的结构和逻辑：

1. db = get_sqlalchemy_db：这行代码获取了一个数据库连接对象。这个对象可能是通过某种方式创建的，比如使用 SQLAlchemy 的 create_engine 函数。
2. try...except...finally：这是一个异常处理结构，用于捕获在数据库操作过程中可能发生的任何异常，并在操作完成后关闭数据库会话。
3. with (Session(db.engine) as session)：这里使用了 SQLAlchemy 的 Session 对象来管理数据库会话。with 语句确保了即使在发生异常的情况下，数据库会话也能被正确关闭。
4. stmt = select(AlchemyTaskManage).where(AlchemyTaskManage.es_index_prefix.isnot(None))：这是构建查询语句的关键部分。select 函数用于指定要查询的表（在这个例子中是 AlchemyTaskManage），而 where 函数用于添加查询条件。这里的条件是 es_index_prefix 字段不为空（isnot(None)）。
5. list_res = session.scalars(stmt).all()：这行代码执行了查询语句，并获取了所有符合条件的记录。scalars 方法用于指定查询结果只包含单个字段的值，而 all 方法用于获取所有符合条件的记录。
6. return [task_info.es_index_prefix for task_info in list_res] if list_res else []：这是一个列表推导式，用于从查询结果中提取 es_index_prefix 字段的值，并返回一个包含这些值的列表。如果查询结果为空，则返回一个空列表。
7. logger.error(f"find task by name An error occurred: {e}")：如果在查询过程中发生异常，这行代码会记录错误信息。
8. db.close_session(session)：在 finally 块中，这行代码确保了数据库会话被关闭，无论查询是否成功。

现在，让我们更深入地探讨如何使用 SQLAlchemy 来查询字段非空的记录。

在 SQLAlchemy 中，查询字段非空通常涉及到使用

isnot(None)

或者

!= None

这样的条件。这是因为在 SQL 中，

NULL

表示没有值，而

None

在 Python 中表示没有值。SQLAlchemy 允许在查询中直接使用

None

来表示 SQL 中的

NULL

。

例如，如果想查询

users

表中

email

字段非空的所有用户，可以这样写：

from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker
from yourapp.models import User

# 创建数据库引擎
engine = create_engine('数据库连接字符串')
Session = sessionmaker(bind=engine)# 创建会话
session = Session()try:# 构建查询语句
    stmt = select(User).where(User.email.isnot(None))# 执行查询
    users_with_email = session.execute(stmt).scalars().all()# 处理查询结果for user in users_with_email:print(user.email)except Exception as e:print(f"An error occurred: {e}")finally:
    session.close()

在这个例子中，我们首先创建了一个数据库引擎和一个会话。然后，我们构建了一个查询语句，指定了要查询的表和条件。接着，我们执行了查询并处理了结果。最后，我们确保了会话被关闭。

在实际应用中，可能需要根据具体的业务逻辑来调整查询条件。例如，可能需要查询某个字段不仅非空，还要满足其他条件，如某个日期字段在特定日期之后，或者某个数值字段在某个范围内。在这种情况下，可以使用

and_

或

or_

函数来组合多个条件。

stmt = select(User).where(User.email.isnot(None), User.last_login >'2024-01-01')

总之，使用 SQLAlchemy 查询字段非空的记录是一种常见的数据库操作。通过合理地构建查询语句和处理查询结果，可以有效地从数据库中提取所需的数据。同时，确保使用异常处理和会话管理来保证数据库操作的安全性和稳定性。

觉得有用的话点个赞

👍🏻

呗。
❤️❤️❤️本人水平有限，如有纰漏，欢迎各位大佬评论批评指正！😄😄😄

💘💘💘如果觉得这篇文对你有帮助的话，也请给个点赞、收藏下吧，非常感谢!👍 👍 👍

🔥🔥🔥Stay Hungry Stay Foolish 道阻且长,行则将至,让我们一起加油吧！🌙🌙🌙