企业官网建设流程全解析

水果网店网站建设策划书,wordpress接收不到邮件,软文推广文章范文1000,邯郸的互联网公司Python MySQL从零上手#xff1a;30分钟搞懂为什么需要ORM 文章目录Python MySQL从零上手#xff1a;30分钟搞懂为什么需要ORM学习开场#xff1a;为什么Python开发者需要关注数据库操作#xff1f;环境准备#xff1a;搭建你的Python MySQL开发环境1. 安装必要的包2. 准备…Python MySQL从零上手30分钟搞懂为什么需要ORM文章目录Python MySQL从零上手30分钟搞懂为什么需要ORM学习开场为什么Python开发者需要关注数据库操作环境准备搭建你的Python MySQL开发环境1. 安装必要的包2. 准备测试数据库基础概念从原始SQL到ORM的演进之路第一节用pymysql实现基础CRUD - 感受原始SQL的痛点第二节为什么我们需要ORM理解核心价值实战演练用SQLAlchemy重构我们的用户管理第三节第一个SQLAlchemy模型怎么写第四节理解SQLAlchemy的架构和工作原理应用场景在实际项目中如何使用SQLAlchemy场景1Web应用中的用户认证系统场景2电商平台的订单管理系统学习总结从原始SQL到ORM的思维转变1. ORM的核心价值2. 什么时候用ORM什么时候用原始SQL3. 我学习ORM时踩过的坑4. 最佳实践建议学习交流与进阶刚开始用Python操作MySQL时我也被SQL注入、连接管理和繁琐的SQL拼接搞得头疼。直到我理解了ORM的价值才发现数据库操作可以如此优雅。今天我就带你从最原始的方式开始一步步理解为什么我们需要SQLAlchemy这样的ORM工具。学习开场为什么Python开发者需要关注数据库操作如果你刚开始用Python做Web开发或数据分析一定会遇到这样的场景用户注册、数据查询、订单管理...这些都需要和数据库打交道。我刚开始做项目时最头疼的就是写SQL语句——拼字符串容易出错防SQL注入要小心翼翼连接管理更是让人头大。你可能会遇到这些问题SQL语句拼错一个引号调试半天找不到原因忘记参数化查询导致SQL注入漏洞连接没有正确关闭数据库连接数很快耗尽数据库表结构改了要手动修改几十处SQL代码别担心这些问题我都经历过。今天我们就从最基础的pymysql开始一步步理解为什么需要ORM以及SQLAlchemy如何让我们的开发更高效。环境准备搭建你的Python MySQL开发环境1. 安装必要的包打开你的终端执行以下命令# 安装MySQL驱动pipinstallpymysql# 安装SQLAlchemyORM框架pipinstallsqlalchemy# 安装mysql-connector-python备用驱动pipinstallmysql-connector-python2. 准备测试数据库我们先创建一个简单的测试数据库。如果你还没有安装MySQL建议使用Docker快速启动# 使用Docker启动MySQLdockerrun-d--namemysql-test\-eMYSQL_ROOT_PASSWORD123456\-eMYSQL_DATABASEtest_db\-p3306:3306\mysql:8.0或者直接在MySQL中执行CREATEDATABASEIFNOTEXISTStest_db;USEtest_db;-- 创建一个用户表CREATETABLEusers(idINTAUTO_INCREMENTPRIMARYKEY,usernameVARCHAR(50)NOTNULLUNIQUE,emailVARCHAR(100)NOTNULL,created_atTIMESTAMPDEFAULTCURRENT_TIMESTAMP);-- 插入一些测试数据INSERTINTOusers(username,email)VALUES(alice,aliceexample.com),(bob,bobexample.com),(charlie,charlieexample.com);基础概念从原始SQL到ORM的演进之路第一节用pymysql实现基础CRUD - 感受原始SQL的痛点让我们先看看不用ORM时我们是怎么操作数据库的importpymysqlfromdatetimeimportdatetime# 1. 建立数据库连接defget_connection():获取数据库连接returnpymysql.connect(hostlocalhost,port3306,userroot,password123456,databasetest_db,charsetutf8mb4)# 2. 查询用户defget_users_raw():使用原始SQL查询用户connNonetry:connget_connection()withconn.cursor()ascursor:# 注意这里容易犯SQL注入的错误sqlSELECT * FROM userscursor.execute(sql)resultscursor.fetchall()# 手动将结果转换为字典列表users[]forrowinresults:user{id:row[0],username:row[1],email:row[2],created_at:row[3]}users.append(user)returnusersexceptExceptionase:print(f查询失败:{e})return[]finally:ifconn:conn.close()# 3. 添加用户容易出错的版本defadd_user_wrong(username,email):❌ 错误示例容易SQL注入connNonetry:connget_connection()withconn.cursor()ascursor:# 危险直接拼接字符串sqlfINSERT INTO users (username, email) VALUES ({username}, {email})cursor.execute(sql)conn.commit()print(添加成功)exceptExceptionase:print(f添加失败:{e})ifconn:conn.rollback()finally:ifconn:conn.close()# 4. 添加用户正确的参数化查询defadd_user_correct(username,email):✅ 正确示例参数化查询connNonetry:connget_connection()withconn.cursor()ascursor:# 使用参数化查询防止SQL注入sqlINSERT INTO users (username, email) VALUES (%s,%s)cursor.execute(sql,(username,email))conn.commit()print(添加成功)exceptExceptionase:print(f添加失败:{e})ifconn:conn.rollback()finally:ifconn:conn.close()# 测试代码if__name____main__:print( 原始SQL操作演示 )# 查询现有用户usersget_users_raw()print(f现有用户:{users})# 测试错误的方式模拟SQL注入print(\n 测试SQL注入风险 )# 如果用户输入是: admin OR 11malicious_inputadmin OR 11add_user_wrong(malicious_input,hackexample.com)# 正确的方式print(\n 使用参数化查询 )add_user_correct(david,davidexample.com)# 再次查询usersget_users_raw()print(f添加后的用户:{users})运行这段代码你会发现几个问题代码冗长每个操作都要写连接管理、异常处理容易出错手动拼接SQL容易出错且不安全结果处理麻烦需要手动将元组转换为字典维护困难如果表结构改了所有SQL都要改第二节为什么我们需要ORM理解核心价值ORMObject-Relational Mapping的核心思想是用操作Python对象的方式来操作数据库。这就像给你的数据库操作加了一个翻译官。ORM解决了什么问题问题类型原始SQL的问题ORM的解决方案SQL注入需要手动参数化容易忘记自动参数化从根本上杜绝代码冗余每个操作都要写连接管理封装通用操作代码复用类型转换手动转换查询结果自动映射为Python对象维护成本表结构改动要改所有SQL只需改模型定义数据库迁移不同数据库SQL语法不同统一接口轻松切换让我用一个生活中的比喻原始SQL就像手动挡汽车你需要自己控制离合、换挡而ORM就像自动挡汽车你只需要告诉它前进或后退复杂的操作它帮你处理。实战演练用SQLAlchemy重构我们的用户管理第三节第一个SQLAlchemy模型怎么写现在让我们用SQLAlchemy重写上面的用户管理功能fromsqlalchemyimportcreate_engine,Column,Integer,String,DateTimefromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportsessionmakerfromdatetimeimportdatetime# 1. 定义基类Basedeclarative_base()# 2. 定义用户模型对应users表classUser(Base):用户模型类__tablename__users# 指定表名# 定义字段idColumn(Integer,primary_keyTrue,autoincrementTrue)usernameColumn(String(50),nullableFalse,uniqueTrue)emailColumn(String(100),nullableFalse)created_atColumn(DateTime,defaultdatetime.now)def__repr__(self):友好的字符串表示returnfUser(id{self.id}, username{self.username}, email{self.email})# 3. 创建数据库引擎和会话definit_database():初始化数据库连接# 创建引擎连接池默认开启enginecreate_engine(mysqlpymysql://root:123456localhost:3306/test_db,echoTrue,# 显示执行的SQL调试用pool_size5,# 连接池大小max_overflow10,# 最大溢出连接数pool_recycle3600# 连接回收时间秒)# 创建所有表如果不存在Base.metadata.create_all(engine)# 创建会话工厂Sessionsessionmaker(bindengine)returnSession# 4. 使用ORM进行CRUD操作classUserManager:用户管理器使用ORMdef__init__(self):self.Sessioninit_database()defget_all_users(self):获取所有用户sessionself.Session()try:# 看这里像操作Python列表一样查询数据库userssession.query(User).all()returnusersfinally:session.close()defadd_user(self,username,email):添加用户sessionself.Session()try:# 创建用户对象new_userUser(usernameusername,emailemail)# 添加到会话session.add(new_user)# 提交事务session.commit()print(f添加成功用户ID:{new_user.id})returnnew_userexceptExceptionase:session.rollback()print(f添加失败:{e})raisefinally:session.close()defget_user_by_username(self,username):根据用户名查询用户sessionself.Session()try:# 使用filter方法自动参数化查询usersession.query(User).filter(User.usernameusername).first()returnuserfinally:session.close()defupdate_user_email(self,username,new_email):更新用户邮箱sessionself.Session()try:# 先查询用户usersession.query(User).filter(User.usernameusername).first()ifuser:# 直接修改对象属性user.emailnew_emailsession.commit()print(f更新成功:{username}-{new_email})returnTrueelse:print(f用户不存在:{username})returnFalseexceptExceptionase:session.rollback()print(f更新失败:{e})returnFalsefinally:session.close()defdelete_user(self,username):删除用户sessionself.Session()try:usersession.query(User).filter(User.usernameusername).first()ifuser:# 删除对象session.delete(user)session.commit()print(f删除成功:{username})returnTrueelse:print(f用户不存在:{username})returnFalseexceptExceptionase:session.rollback()print(f删除失败:{e})returnFalsefinally:session.close()# 5. 测试ORM版本if__name____main__:print( SQLAlchemy ORM操作演示 )managerUserManager()# 查询所有用户print(\n1. 查询所有用户:)usersmanager.get_all_users()foruserinusers:print(f{user})# 添加新用户print(\n2. 添加新用户:)try:new_usermanager.add_user(emma,emmaexample.com)print(f 添加的用户:{new_user})exceptExceptionase:print(f 添加时出错可能是用户名重复:{e})# 查询特定用户print(\n3. 查询特定用户:)usermanager.get_user_by_username(alice)print(f 查询结果:{user})# 更新用户print(\n4. 更新用户邮箱:)manager.update_user_email(bob,bob.newexample.com)# 再次查询验证更新bobmanager.get_user_by_username(bob)print(f 更新后的Bob:{bob})# 删除用户print(\n5. 删除用户:)manager.delete_user(charlie)# 最终用户列表print(\n6. 最终用户列表:)final_usersmanager.get_all_users()foruserinfinal_users:print(f{user})运行这段代码你会看到SQLAlchemy自动生成的SQL语句因为设置了echoTrue。对比一下两种方式的区别操作原始SQL代码行数ORM代码行数可读性对比查询所有用户15行含连接管理3行ORM更简洁添加用户20行含异常处理5行ORM更直观条件查询需要写完整SQL链式调用更PythonicORM胜出更新数据UPDATE语句WHERE直接修改对象属性ORM更自然防止注入需手动参数化自动处理ORM更安全第四节理解SQLAlchemy的架构和工作原理SQLAlchemy不是简单的SQL生成器它是一个完整的ORM框架包含多个组件SQLAlchemy的核心组件Engine引擎数据库连接的工厂管理连接池Session会话对象容器管理事务状态Model模型数据库表的Python类表示Query查询构建和执行查询的接口一个查询的完整旅程应用场景在实际项目中如何使用SQLAlchemy场景1Web应用中的用户认证系统假设我们在开发一个Flask Web应用需要用户注册登录功能# web_app/models.py - 数据模型定义fromsqlalchemyimportColumn,Integer,String,Booleanfromsqlalchemy.ext.declarativeimportdeclarative_basefromwerkzeug.securityimportgenerate_password_hash,check_password_hashBasedeclarative_base()classWebUser(Base):__tablename__web_usersidColumn(Integer,primary_keyTrue)usernameColumn(String(64),uniqueTrue,nullableFalse)emailColumn(String(120),uniqueTrue,nullableFalse)password_hashColumn(String(256),nullableFalse)is_activeColumn(Boolean,defaultTrue)is_adminColumn(Boolean,defaultFalse)defset_password(self,password):设置密码自动哈希self.password_hashgenerate_password_hash(password)defcheck_password(self,password):验证密码returncheck_password_hash(self.password_hash,password)classmethoddefauthenticate(cls,session,username,password):用户认证usersession.query(cls).filter((cls.usernameusername)|(cls.emailusername)).first()ifuseranduser.check_password(password):returnuserreturnNone# web_app/auth.py - 认证逻辑fromsqlalchemy.ormimportSessiondefregister_user(session:Session,username,email,password):用户注册# 检查用户是否已存在existingsession.query(WebUser).filter((WebUser.usernameusername)|(WebUser.emailemail)).first()ifexisting:returnNone,用户名或邮箱已存在# 创建新用户new_userWebUser(usernameusername,emailemail)new_user.set_password(password)session.add(new_user)session.commit()returnnew_user,注册成功deflogin_user(session:Session,username,password):用户登录userWebUser.authenticate(session,username,password)ifuser:ifnotuser.is_active:returnNone,账户已被禁用returnuser,登录成功returnNone,用户名或密码错误场景2电商平台的订单管理系统# ecommerce/models.pyfromsqlalchemyimportColumn,Integer,String,Float,DateTime,ForeignKeyfromsqlalchemy.ormimportrelationshipfromdatetimeimportdatetimeclassProduct(Base):__tablename__productsidColumn(Integer,primary_keyTrue)nameColumn(String(100),nullableFalse)priceColumn(Float,nullableFalse)stockColumn(Integer,default0)# 定义关系一对多order_itemsrelationship(OrderItem,back_populatesproduct)defdecrease_stock(self,quantity):减少库存ifself.stockquantity:self.stock-quantityreturnTruereturnFalseclassOrder(Base):__tablename__ordersidColumn(Integer,primary_keyTrue)user_idColumn(Integer,ForeignKey(web_users.id))total_amountColumn(Float,default0.0)statusColumn(String(20),defaultpending)# pending, paid, shipped, completedcreated_atColumn(DateTime,defaultdatetime.now)# 定义关系userrelationship(WebUser,backreforders)itemsrelationship(OrderItem,back_populatesorder)defcalculate_total(self):计算订单总金额self.total_amountsum(item.subtotalforiteminself.items)returnself.total_amountclassOrderItem(Base):__tablename__order_itemsidColumn(Integer,primary_keyTrue)order_idColumn(Integer,ForeignKey(orders.id))product_idColumn(Integer,ForeignKey(products.id))quantityColumn(Integer,nullableFalse)priceColumn(Float,nullableFalse)# 下单时的价格# 计算字段不存储动态计算propertydefsubtotal(self):returnself.quantity*self.price# 定义关系orderrelationship(Order,back_populatesitems)productrelationship(Product,back_populatesorder_items)# ecommerce/services.py - 业务逻辑classOrderService:订单服务def__init__(self,session):self.sessionsessiondefcreate_order(self,user_id,items):创建订单try:# 开启事务orderOrder(user_iduser_id)self.session.add(order)foritem_datainitems:product_iditem_data[product_id]quantityitem_data[quantity]# 查询商品productself.session.query(Product).get(product_id)ifnotproduct:raiseValueError(f商品不存在:{product_id})# 检查库存ifnotproduct.decrease_stock(quantity):raiseValueError(f库存不足:{product.name})# 创建订单项order_itemOrderItem(orderorder,productproduct,quantityquantity,priceproduct.price)self.session.add(order_item)# 计算总金额order.calculate_total()# 提交事务self.session.commit()returnorderexceptExceptionase:self.session.rollback()raise学习总结从原始SQL到ORM的思维转变通过今天的学习你应该理解了1. ORM的核心价值开发效率用Pythonic的方式操作数据库代码安全自动防止SQL注入维护便利表结构变更影响最小化可移植性轻松切换数据库2. 什么时候用ORM什么时候用原始SQL场景推荐方式理由日常CRUD操作ORM开发快维护易复杂报表查询原始SQL或混合SQL更直观优化更方便批量数据导入原始SQL性能更高原型开发ORM快速迭代性能关键路径根据情况选择需要实际测试3. 我学习ORM时踩过的坑N1查询问题早期不知道.joinedload()一个列表查询变成几十个查询会话管理混乱多个地方创建session导致事务不一致过度查询.all()加载全部数据内存爆炸忽略索引ORM方便了但忘记给常用查询字段加索引4. 最佳实践建议# ✅ 好的做法defget_users_good(session):# 使用joinedload避免N1查询userssession.query(User).options(joinedload(User.orders)).filter(User.is_activeTrue).all()returnusers# ❌ 不好的做法defget_users_bad(session):userssession.query(User).filter(User.is_activeTrue).all()foruserinusers:# 每次循环都查询数据库N1问题orderssession.query(Order).filter(Order.user_iduser.id).all()user.ordersordersreturnusers学习交流与进阶恭喜你完成了SQLAlchemy ORM的入门学习现在你应该理解了为什么我们需要ORM以及它如何让我们的数据库操作更安全、更高效。欢迎在评论区分享你在学习ORM时遇到了哪些困惑文中的示例代码运行成功了吗在实际项目中你更倾向于用ORM还是原始SQL为什么我会认真阅读每一条留言并为初学者提供针对性的解答。记得多动手实践这是掌握数据库编程最好的方式推荐学习资源SQLAlchemy官方文档 - 最权威的学习资料从入门到精通SQLAlchemy教程 by Miguel Grinberg - 实战性很强的教程Alembic数据库迁移指南 - 学习如何管理数据库变更下篇预告下一篇将分享《手把手搞定SQLAlchemy ORM》带你搞定SQLAlchemy ORM从模型定义到MySQL实战映射。学习建议数据库学习就像学游泳光看教程不下水永远学不会。建议你把今天的代码都敲一遍尝试修改表结构体验ORM的便利设计一个自己的小项目如博客系统、TODO应用遇到问题不要怕每个开发者都是从错误中成长起来的。加油