flask 5 wtforms实现流程、sqlalchemy相关操作、面向对象补充、面试相关
2021/4/26 19:27:27
本文主要是介绍flask 5 wtforms实现流程、sqlalchemy相关操作、面向对象补充、面试相关,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!
Flask day 5 wtforms实现流程、sqlalchemy相关操作、面向对象补充、面试相关
内容回顾
1.flask和django区别 -相同: -基于wsgi -不同 -传值方式 -组件 -flask少 -django多 2.flask上下文管理是如何实现的? -流程: -请求刚进来,RequestContext(request,session),AppContext(app,g)——> LocalStack(将数据维护成一个栈)——>Local -视图处理,LocalProxy——>偏函数——>LocalStack——>Local -请求结束,save_session——>LocalStack.pop() 3.Local的作用? -用于保存 -请求上下文对象 -app上下文 -并且可以做到`线程`间的数据隔离 线程:threading.local 协程:greenlet.get_current as get_ident 4.LocalStack? -将Local中保存的数据维护成一个栈 { 1234:{stact:[ctx,]} } 5.Flask内置组件 -配置 -路由 -视图 -模板 -session -蓝图 -闪现 -装饰器 -中间件 6.第三方组件: -Flask: -flask-session,将原来保存在cookie中的session数据,放到redis/memcache/文件/数据库中 -公共: -dbutils,数据库连接池 -wtforms,用于做表单验证+生成HTML标签
今日内容
1.面向对象相关
1.1 mro
c3算法找到当前类寻找属性的顺序
1.2 创建类的两种方式
方式一: class Foo: city = 'sh' def func(self,x): return x + 1 方式二: Foo = type('Foo',(object,),{'city':'sh'}, 'func':lambda self,x:x+1)
1.3 类由自定义的type创建
class MyType(type): def __init__(self, *args, **kwargs): print("创建类之前") super().__init__(*args, **kwargs) print("创建类之后") class Foo(object, metaclass=MyType): # 当前类,由type类创建 city = 'sh' def func(self, x): return x + 1 >>>>>>>> 创建类之前 创建类之后
1.4 类的继承
类的基类中指定了metaclass,那么当前类也是由metaclass指定的类来创建当前类。
class MyType(type): def __init__(self, *args, **kwargs): print("创建类之前") super().__init__(*args, **kwargs) print("创建类之后") class Foo(object, metaclass=MyType): # 当前类,由type类创建 city = 'sh' def func(self, x): return x + 1 class Bar(Foo): pass >>>>>>>> 创建类之前 创建类之后 创建类之前 创建类之后
1.5 with_metaclass
def with_metaclass(meta, base=object): return meta("NewBase", (base,), {}) class Form(with_metaclass(FormMeta, BaseForm)): pass FormMeta("NewBase", (BaseForm,), {}) class NewBase(BaseForm, metaclass=FormMeta): pass class Form(NewBase): pass
2.wtforms
class FormMeta(type): def __init__(cls, name, bases, attrs): type.__init__(cls, name, bases, attrs) cls._unbound_fields = None cls._wtforms_meta = None class NewBase(BaseForm, metaclass=FormMeta): pass class Form(NewBase): pass class RegisterForm(Form): pass # 所有在定义RegisterForm类的时候就有了 RegisterForm._unbound_fields = None RegisterForm._wtforms_meta = None
wtforms实现流程
1.1 实例化流程
class RegisterForm(Form): name = simple.StringField(正则=[验证规则1,验证规则2,],插件=Input框) pwd = simple.PasswordField(正则=[验证规则1,验证规则2,],插件=Input框) 创建类的初始化过程中会有 """ RegisterForm._unbound_fields = None RegisterForm._wtforms_meta = None RegisterForm.name = UnboundField(simple.StringField, *args, **kwargs, creation_counter=0,_formfield=True) RegisterForm.pwd = UnboundField(simple.PasswordField, *args, **kwargs, creation_counter=1,_formfield=True) """ form = RegisterForm() """ 该过程会依次执行 FormMeta.__call__ RegisterForm.__new__ RegisterForm.__init__ FormMeta.__call__ """ def __call__(cls, *args, **kwargs): if cls._unbound_fields is None: fields = [] # 实例化字典中的key值: _unbound_fields,_wtforms_meta,xxx,name,pwd... for name in dir(cls): # xxx,name,pwd if not name.startswith('_'): # 获取对应key的值相当于cls.xxx,cls.name,cls.pwd unbound_field = getattr(cls, name) # UnboundField对象中有_formfield=True if hasattr(unbound_field, '_formfield'): fields.append((name, unbound_field)) # 此时fields = [ ("name",UnboundField(simple.StringField, *args, **kwargs, creation_counter=0,_formfield=True)), ("pwd",UnboundField(simple.PasswordField, *args, **kwargs, creation_counter=1,_formfield=True)) ] # 排序 fields中的字段让creation_counter从小到大排序 fields.sort(key=lambda x: (x[1].creation_counter, x[0])) cls._unbound_fields = fields if cls._wtforms_meta is None: bases = [] # __mro__= [RegisterForm, Form, BaseForm, object] for mro_class in cls.__mro__: # 只有Form中有Meta if 'Meta' in mro_class.__dict__: bases.append(mro_class.Meta) # 此时bases=[ # 如果自己定义了Meta类:RegisterForm.Meta类, # Form.Meta=DefaultMeta类, ] # class Meta(Meta类,DefaultMeta类):pass cls._wtforms_meta = type('Meta', tuple(bases), {}) return type.__call__(cls, *args, **kwargs) # 结束之后 RegisterForm._unbound_fields = [ ("name",UnboundField(simple.StringField, *args, **kwargs, creation_counter=0,_formfield=True)), ("pwd",UnboundField(simple.PasswordField, *args, **kwargs, creation_counter=1,_formfield=True)) ] RegisterForm._wtforms_meta = class Meta(DefaultMeta):pass RegisterForm.name = UnboundField(simple.StringField, *args, **kwargs, creation_counter=0,_formfield=True) RegisterForm.pwd = UnboundField(simple.PasswordField, *args, **kwargs, creation_counter=1,_formfield=True) # 接着执行 RegisterForm.__new__ pass # 最后执行 RegisterForm.__init__ RegisterForm对象._fields = { 'name':simple.StringField(实例化), 'pwd':simple.PasswordField() } RegisterForm对象.setattr(self,name,val) RegisterForm对象.name = simple.StringField() RegisterForm对象.pwd = simple.PasswordField() # form = RegisterForm(data={'name': 'lem'}) # 给input框添加value默认值 print(form.name) # 执行simple.StringField().__str__ ,希望生成input框 # 执行simple.StringField().__call__ # 调用meta.render_field(self,kwargs) # - simple.StringField().widgets() # - TextInput.__call__
1.2验证流程(下一节)
1.3相关面试题
a. Python基础部分,哪些比较重要? -反射 - CVB - jdango 配置文件 - wtforms class Form(with_metaclass(FormMeta, BaseForm)): Meta = DefaultMeta def __init__(self, formdata=None, obj=None, prefix='', data=None, meta=None, **kwargs): meta_obj = self._wtforms_meta() if meta is not None and isinstance(meta, dict): meta_obj.update_values(meta) super(Form, self).__init__(self._unbound_fields, meta=meta_obj, prefix=prefix) for name, field in iteritems(self._fields): setattr(self, name, field) self.process(formdata, obj, data=data, **kwargs) -装饰器 - flask路由/装饰器 - 认证 - csrf -生成器、迭代器 -面向对象 -继承、封装、多态 -特殊功能: -双下划线方法 - __mro__ - wtforms中定义Meta - __dict__ - dir - dict - __new__,实例化但是没有给当前对象 -wtforms,字段实例化时返回,不是StringField,而是UnboundField - rest framework, many=True - __call__ - flask请求的入口 - 字段生成标签时:字段.__str__=>字段.__call__=>字段.__call__ - __iter__:让实例化的对象可以被for循环 -metaclass -作用:用于指定使用哪个类来创建当前类 -场景:在类创建之前定制操作 示例:在wtforms中,对字段进行排序。
3.SQLALchemy/flask-sqlalchemy
3.1 ORM
ORM框架 问题:什么是ORM?(Object Relational Mapping) -关系对象映射 类 ——>表 对象——>记录(一行数据) 当有了对应关系之后,不再需要编写SQL语句,取而代之的是操作类和对象 -ORM和原生SQL哪个好? -ORM开发快编写简单 -SQL运行效率高 -概念 -db first:根据数据库的表生成类 django:python manage.py inspectdb - code first:根据类创建数据库表 django:python manage.py makemigrations python manage.py migrate -ORM是怎么实现? DDD中: unit of work
3.2 SQLALchemy
sqlalchemy是一个基于python实现ORM框架
from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, UniqueConstraint, Index from sqlalchemy import create_engine Base = declarative_base() # 创建表单 class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(32)) extra = Column(String(32)) # 数据库连接相关 engine = create_engine("mysql+pymysql://root:123456@localhost:3306/orm_test") # 创建表 Base.metadata.create_all(engine) # 删除表 Base.metadata.drop_all(engine)
基于ORM使用
from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker import models engine = create_engine("mysql+pymysql://root:123456@localhost:3306/orm_test") Session = sessionmaker(bind=engine) session = Session() obj1 = models.User(name='lem', extra='lp') obj2 = models.User(name='lem', extra='wife') session.add(obj1) session.add(obj2) session.commit()
基于原生MYSQL使用以及sqlalchemy连接池
from sqlalchemy import create_engine import asyncio import time from threading import Thread engine = create_engine("mysql+pymysql://root:123456@localhost:3306/orm_test", max_overflow=0, # 超过连接池大小外最多创建的链接 pool_size=10, # 连接池大小 pool_timeout=30, # 池中没有链接,线程最多等待多少时间,否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置) ) # cur = engine.execute("select * from users") # 获取第一行数据 # print(cur.fetchone()) # 获取n行数据 # print(cur.fetchmany(2)) # 获取所有数据 # print(cur.fetchall()) def task(i): print("开始task", i) conn = engine.raw_connection() cursor = conn.cursor() cursor.execute("select * from users") res = cursor.fetchall() cursor.close() conn.close() print("task结束", i, res) return res async def get_info(i): loop = asyncio.get_event_loop() fut = loop.run_in_executor(None, task, i) res = await fut if __name__ == '__main__': t1 = time.time() tasks = [get_info(i) for i in range(20)] asyncio.run(asyncio.wait(tasks)) # ls = [] # for i in range(20): # t = Thread(target=task, args=(i,)) # t.start() # ls.append(t) # # for i in ls: # i.join() print(time.time()-t1) >>>>>>>> 2.0804262161254883 >>>>>>>> 2.091407060623169
增删改查
import time import threading from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index from sqlalchemy.orm import sessionmaker, relationship from sqlalchemy import create_engine from sqlalchemy.sql import text from db import Users, Hosts engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5) Session = sessionmaker(bind=engine) session = Session() # ################ 添加 ################ """ obj1 = Users(name="wupeiqi") session.add(obj1) session.add_all([ Users(name="wupeiqi"), Users(name="alex"), Hosts(name="c1.com"), ]) session.commit() """ # ################ 删除 ################ """ session.query(Users).filter(Users.id > 2).delete() session.commit() """ # ################ 修改 ################ """ session.query(Users).filter(Users.id > 0).update({"name" : "099"}) session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False) session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate") session.commit() """ # ################ 查询 ################ """ r1 = session.query(Users).all() r2 = session.query(Users.name.label('xx'), Users.age).all() r3 = session.query(Users).filter(Users.name == "alex").all() r4 = session.query(Users).filter_by(name='alex').all() r5 = session.query(Users).filter_by(name='alex').first() r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(Users.id).all() r7 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed').all() """ session.close()
常用操作
# 条件 ret = session.query(Users).filter_by(name='alex').all() ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all() ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all() ret = session.query(Users).filter(Users.id.in_([1,3,4])).all() ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all() ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all() from sqlalchemy import and_, or_ ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all() ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all() ret = session.query(Users).filter( or_( Users.id < 2, and_(Users.name == 'eric', Users.id > 3), Users.extra != "" )).all() # 通配符 ret = session.query(Users).filter(Users.name.like('e%')).all() ret = session.query(Users).filter(~Users.name.like('e%')).all() # 限制 ret = session.query(Users)[1:2] # 排序 ret = session.query(Users).order_by(Users.name.desc()).all() ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all() # 分组 from sqlalchemy.sql import func ret = session.query(Users).group_by(Users.extra).all() ret = session.query( func.max(Users.id), func.sum(Users.id), func.min(Users.id)).group_by(Users.name).all() ret = session.query( func.max(Users.id), func.sum(Users.id), func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all() # 连表 ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all() ret = session.query(Person).join(Favor).all() ret = session.query(Person).join(Favor, isouter=True).all() # 组合 q1 = session.query(Users.name).filter(Users.id > 2) q2 = session.query(Favor.caption).filter(Favor.nid < 2) ret = q1.union(q2).all() q1 = session.query(Users.name).filter(Users.id > 2) q2 = session.query(Favor.caption).filter(Favor.nid < 2) ret = q1.union_all(q2).all() 常用操作
基于relationship操作foreignkey
#!/usr/bin/env python # -*- coding:utf-8 -*- import time import threading from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index from sqlalchemy.orm import sessionmaker, relationship from sqlalchemy import create_engine from sqlalchemy.sql import text from sqlalchemy.engine.result import ResultProxy from db import Users, Hosts, Hobby, Person engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5) Session = sessionmaker(bind=engine) session = Session() # 添加 """ session.add_all([ Hobby(caption='乒乓球'), Hobby(caption='羽毛球'), Person(name='张三', hobby_id=3), Person(name='李四', hobby_id=4), ]) person = Person(name='张九', hobby=Hobby(caption='姑娘')) session.add(person) hb = Hobby(caption='人妖') hb.pers = [Person(name='文飞'), Person(name='博雅')] session.add(hb) session.commit() """ # 使用relationship正向查询 """ v = session.query(Person).first() print(v.name) print(v.hobby.caption) """ # 使用relationship反向查询 """ v = session.query(Hobby).first() print(v.caption) print(v.pers) """ session.close() 基于relationship操作ForeignKey
面试相关
1.python和其他语言的区别 2.为什么要学python 3.数据类型及用法 4.函数 -函数参数传递的是什么?-引用 -列表生成式 [] -生成器表达式 (for i in range(4)) -题: val = [lambda :i+1 for i in range(4)] print(val[0]) print(val[0]()) print(val[1]()) 其中val等价于下面的ls ls = [] for i in range(4): def func(): return i ls.append(func) print(i) print(ls[0]) print(ls[0]()) # 也可以把函数定义在外面,函数只有在调用时才会触发函数体内代码 此时i在经历for循环后已经变为3 # 所以ls中所有函数return的都是3 5.生成器、迭代器 -迭代器:内置有__iter__和__next__的对象,可迭代对象是内置有__iter__方法的,迭代器也是可迭代对象 -生成器:是一个以yeild作为返回关键字,返回迭代器的函数 -区别: 生成器能做到迭代器能做的所有事,而且自动创建了 iter()和 next()方法。除了创建和保存程序状态的自动方法,当生成器终结时,还会自动抛出 StopIteration 异常。 生成器不要求你事先准备好整个迭代过程中所有的元素,即无须将对象的所有元素都存入内存之后,才开始进行操作。仅在迭代至某个元素时才会将该元素放入内存,而在这之前或之后,元素可以不存在或者被销毁。这个特点使得它特别适合用于遍历一些巨大的或是无限的类序列对象 6.常见内置函数 -map -reduce -filter -zip -instance -type
这篇关于flask 5 wtforms实现流程、sqlalchemy相关操作、面向对象补充、面试相关的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!
- 2024-11-01UniApp 中组件的生命周期是多少-icode9专业技术文章分享
- 2024-11-01如何使用Svg Sprite Icon简化网页图标管理
- 2024-10-31Excel数据导出课程:新手从入门到精通的实用教程
- 2024-10-31Excel数据导入课程:新手入门指南
- 2024-10-31RBAC的权限课程:新手入门教程
- 2024-10-31Svg Sprite Icon课程:新手入门必备指南
- 2024-10-31怎么配置 L2TP 允许多用户连接-icode9专业技术文章分享
- 2024-10-31怎么在FreeBSD上 安装 OpenResty-icode9专业技术文章分享
- 2024-10-31运行 modprobe l2tp_ppp 时收到“module not found”消息提醒是什么-icode9专业技术文章分享
- 2024-10-31FreeBSD的下载命令有哪些-icode9专业技术文章分享