Python优雅编程：会报恩的代码（二）

吴功勋

文章目录引言defaultdictrandomnetworkx固定点位置logging.basicConfig使用SQLAlchemy执行任意SQL安装SQLAlchemy创建SQLAlchemy引擎执行任意SQL使用事务使用db.sesstion.execute执行SQL结语引言在Python编程中，有一些常用技巧和最佳实践可以帮助你编写更优雅、更高效的代码。本文将介绍的是Python中defaultdict的分组作用、random在随机选取中的实践、使用networkx作图时，如何令图中的点一直保持相同的位置（不需要自行指定点的位置）、logging.basicConfig的参数含义、SQLAlchemy执行任意SQL等。defaultdict在Python中，可以使用collections.defaultdict来根据字典的值对条目进行分组。下面是一个示例代码，展示了如何实现这一点：fromcollectionsimportdefaultdict#假设字典名为my_dictmy_dict={'a':1,'b':2,'c':1,'d':3,'e':2}#创建一个默认字典以进行分组grouped_dict=defaultdict(list)#遍历字典并根据值进行分组forkey,valueinmy_dict.items():grouped_dict[value].append(key)#将defaultdict转换为普通字典（可选）result=dict(grouped_dict)print(result)12345678910111213141516171819202122在这个示例中，result将显示分组后的字典，类似于：{1:['a','c'],2:['b','e'],3:['d']}12345每个键是原始字典的值，而对应的值是具有相同值的键的列表。random在Python中，可以使用random模块中的sample函数从列表中随机选择指定数量的对象。以下是如何选择20个对象的示例代码：importrandom#假设列表名为my_listmy_list=[...]#列表内容#从列表中随机选择20个对象random_selection=random.sample(my_list,20)print(random_selection)123456789注意事项：my_list需要至少包含20个元素，否则会引发错误。sample函数不会修改原始列表。networkx固定点位置在使用networkx绘制图形时，可以通过设置node的位置属性来固定节点的位置。可以使用spring_layout、circular_layout、random_layout等布局方法来自动计算节点的位置，然后再保存这些位置，以便后续的绘制中使用。以下是一个示例：importnetworkxasnximportmatplotlib.pyplotasplt#创建一个示例图G=nx.Graph()G.add_edges_from([(1,2),(1,3),(2,4),(3,4)])#使用springlayout计算节点的位置pos=nx.spring_layout(G)#绘制图形并固定节点位置nx.draw(G,pos,with_labels=True)#进行第一次绘制plt.show()#进行其他绘制，仍然使用已保存的位置nx.draw(G,pos,with_labels=True,node_color='lightblue')#再次绘制图形plt.show()123456789101112131415161718192021在这个示例中，节点的位置在第一次绘制时计算并存储在pos变量中。之后，您可以多次使用这个位置绘制图形，而不需要重新计算节点的位置。这样可以确保节点位置保持一致。logging.basicConfig解释如下logging.basicConfig配置的作用。logging.basicConfig(level=app.config.get('LOG_LEVEL'),format=app.config.get('LOG_FORMAT'),datefmt=app.config.get('LOG_DATEFORMAT'),handlers=log_handlers,force=True)1234567这段代码使用Python默认的logging模块的basicConfig函数来配置日志记录的行为。代码的具体说明如下：logging.basicConfig：这是设置日志记录的基本配置的函数。它的一些参数用来控制日志显示的级别、格式以及其他记录方式。level=app.config.get('LOG_LEVEL')：这行代码设置日志记录器的最低日志级别。只有达到这个级别或更高级别的日志才会被记录。如DEBUG、INFO、WARNING、ERROR、CRITICAL等。这里通过app.config.get('LOG_LEVEL')获取配置中的日志级别。format=app.config.get('LOG_FORMAT')：这个参数定义了日志消息的格式。通过app.config.get('LOG_FORMAT')获取配置的日志格式字符串，格式可以包含时间戳、日志级别、消息等信息，通常使用占位符（如%(asctime)s）来定义。datefmt=app.config.get('LOG_DATEFORMAT')：datefmt参数用来指定日志中时间戳的格式。通过app.config.get('LOG_DATEFORMAT')获取配置中的日期格式字符串。它可以使用标准的日期格式来格式化输出的时间。handlers=log_handlers：handlers参数用于定义一个或多个日志处理器（handlers），这些处理器决定了日志信息将被输出到何处（如文件、控制台等）。log_handlers变量应该是在代码的其他部分定义的，包含了一个或多个有效的日志处理器实例。force=True：这个参数用于强制重新配置日志系统。如果日志系统已经配置过，设置force=True将允许使用新的配置。同时，原有的配置将被覆盖。这段代码的目的在于根据应用配置动态设置日志记录的级别、格式、时间格式和处理器，确保日志被正确记录和输出。这种灵活配置常用于应用开发中，以便根据环境（如开发、测试、生产）自动调整日志设置。使用SQLAlchemy执行任意SQL使用SQLAlchemy执行任意SQL查询是相对简单的。以下是一个基本的步骤指南，包括一个示例。我们将创建一个数据库连接，并使用execute方法来执行原始SQL语句。安装SQLAlchemy首先，确保已经安装了SQLAlchemy。可以通过以下命令安装：pipinstallsqlalchemy1如果使用的是数据库特定的驱动（例如SQLite、PostgreSQL或MySQL），也需要安装对应的库，如psycopg2（PostgreSQL）或mysqlclient（MySQL）。创建SQLAlchemy引擎使用SQLAlchemy创建一个引擎连接到数据库。以下是一个示例：fromsqlalchemyimportcreate_engine#替换为数据库连接字符串engine=create_engine('sqlite:///example.db')#这里以SQLite为例1234执行任意SQL使用engine对象的execute方法来执行任意SQL查询。以下是具体示例：#创建连接withengine.connect()asconnection:#执行任意SQL查询result=connection.execute("SELECT*FROMyour_table_name")#替换为SQL查询#处理结果forrowinresult:print(row)#逐行访问查询结果12345678使用事务如果SQL操作是插入、更新或删除，建议使用事务。示例：fromsqlalchemyimporttext#创建连接并使用事务withengine.connect()asconnection:withconnection.begin():#开始一个事务connection.execute(text("INSERTINTOyour_table_name(column1,column2)VALUES('value1','value2')"))#其他SQL操作...1234567注意事项SQL注入：当使用原始SQL时，不要将用户输入直接纳入查询，避免SQL注入。使用参数化查询是良好的实践，如下所示：#使用参数化查询name='value1'withengine.connect()asconnection:connection.execute(text("SELECT*FROMyour_table_nameWHEREcolumn1=:name"),{'name':name})1234关闭连接：当不再需要数据库连接时，务必关闭连接，虽然在使用with语句时，连接会自动关闭。通过上述步骤可以使用SQLAlchemy执行任意SQL查询，并且能够高效地处理结果和事务。确保在执行SQL时注意安全问题，尤其是防止SQL注入。使用db.sesstion.execute执行SQL在SQLAlchemy中，使用db.session.execute来执行SQL查询后，可以通过不同的方法获取执行结果。这些方法的选择取决于查询类型（如SELECT、INSERT、UPDATE等）。首先，确保数据库会话db.session已经设置好。以下是一个基本示例，展示如何使用db.session.execute执行SQL查询并获取结果。执行SELECT查询如果执行的是SELECT查询，可以使用scalars()或fetchall()方法来获取结果。fromsqlalchemyimporttext#假设db是SQLAlchemy实例sql_query=text("SELECT*FROMyour_table_name")#替换为实际的SQL查询withdb.session.begin():#开始一个事务，确保会话在有效范围内result=db.session.execute(sql_query)#获取所有行rows=result.fetchall()forrowinrows:print(row)#输出每一行#获取单一列的结果（如果只需要某一列）single_column=result.scalars().all()#获取所有单列结果print(single_column)1234567891011121314151617执行INSERT,UPDATE,DELETE查询对于INSERT、UPDATE或DELETE语句，可以通过rowcount属性获取受影响的行数。sql_insert=text("INSERTINTOyour_table_name(column1,column2)VALUES(:value1,:value2)")withdb.session.begin():result=db.session.execute(sql_insert,{'value1':'data1','value2':'data2'})print(result.rowcount)#输出受影响的行数12345注意事项使用text()来包裹SQL查询，以确保SQLAlchemy可以正确解析。确保在使用db.session的地方有正确的事务管理（如使用withdb.session.begin()）。如果SQL查询返回的是多个列，可以直接通过result.fetchall()获取所有行，同时每一行都是一个元组。对于数据写入操作（INSERT、UPDATE、DELETE），通常只关心操作成功与否和受影响的行数。使用db.session.execute可以轻松执行SQL查询并获取结果。对于SELECT查询，可以使用fetchall()或scalars()方法获取结果，而对于数据修改操作，则可以通过rowcount来查看影响的行数。确保使用text()以保证SQL语句被正确处理和解析。结语本文介绍了Python中的defaultdict、random、networkx、logging、SQLAlchemy的一些特定用法，希望这些小技巧能在某个特定的时间正好帮到你。PS：感谢每一位志同道合者的阅读，欢迎关注、点赞、评论！上一篇：Python优雅编程：会报恩的代码（一）专栏：「计算通践」