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
标签：__,sqlalchemy,name,flask,session,wtforms,import,query,Users 来源： https://www.cnblogs.com/HinaChan/p/14705061.html