码厩开发者笔记

#22 Python 常见后缀

.gitignore 常见的 *.py[cod]。

参考 gitignore.io。

• .py
• .pyc 编译过后生成的字节码文件
• .pyo 经过优化的字节码文件。Python2 时代常见，Python3 不再使用。现在是 .opt-1.pyc, opt-2.pyc
  PS：关于这个优化到底是做什么以后再好好研究。
• .pyd
• .pyi Python Stub 文件，IDE 中为了提供更好的提示会带一些这样的文件，只是所有方法都是空的。
  ps: 现在的 PyCharm 好像带的 stub 文件都是 py 后缀了。
• .pyw Windows 专供，py 文件由 python.exe 解析，pyw 文件由 pythonw.exe 执行
  他们不同之处就是，python.exe 会导致一个命令行窗口的出现，而，pythonw.exe 不会，所以常用在 GUI 程序上。
• .pth 自定义路径导入规则的文件，参考：.pth 文件和 site 模块

#21 Python 文件压缩方式汇总

• zlib — Compression compatible with gzip
• gzip — Support for gzip files
• bz2 — Support for bzip2 compression
• lzma — Compression using the LZMA algorithm
• zipfile — Work with ZIP archives
• tarfile — Read and write tar archive files

#20 安装和使用 pep8

pep8 是一个遵守 PEP8 编码规范的 Python 代码检测工具。

#19 Tornado & HTTP 599

Tornado POST 请求直接 599 了，经过 WireShark 抓包发现根本没有发出请求。
最后竟然发现和版本有关，4.1 下才有问题，4.2 以后就好了。

#18 单文件启动 Django 应用

没有什么实际意义，只是玩玩而已。

#17 Unicode 编码中中文字符的范围

突然来了兴致，想看看 Unicode 中有多少个中文，查了一下，很多人都是说 4e00 至 9fff 段¹。

# -*- coding: utf-8 -*-

all_chinese_in_unicode = range(0x4e00, 0x9fff)


def transfer(u_char_num):
    if isinstance(u_char_num, int):
        u_char_hex = '%x' % u_char_num
        u_char_str = '\u' + u_char_hex
    else:
        if isinstance(u_char_num, str) and len(u_char_num) == 4:
            u_char_str = '\u' + u_char_num
        else:
            raise Exception
    u_char = u_char_str.decode('raw_unicode_escape')
    # print u_char_hex, u_char
    return u_char


def test_transfer():
    # repr(u"国") -> u'\u56fd' -> 22269
    print transfer(0x56fd)


# 打印所有中文字符
# for i in all_chinese_in_unicode:
#    print transfer(i),
# print

# 打印最后一个中文字符及前、后各一个字符
last_chinese_char = 0x9fbb
last_chinese_char_index = all_chinese_in_unicode.index(last_chinese_char)
start, end = last_chinese_char_index - 1, last_chinese_char_index + 2
for i in all_chinese_in_unicode[start:end]:
    print transfer(i),
print

Ubuntu 下的 zsh 中运行，只能显示到这个字符：龻，后面的都是乱码，这个字符对应的十六进制数是 9fbb。

结果又意外发现，最后一个字符似乎不是 9fbb，而是 9fcc（改 URL 一个一个试出来的）。

来源：https://www.fileformat.info/info/unicode/char/9fcc/index.htm
对应文字图片：https://www.fileformat.info/info/unicode/char/9fcc/sample.png

当然，这个来源并不保证权威，也可能有错误，涉及中文字符的范围还是使用 4e00 - 9fff 比较保险。

比如正则表达式：[\u4e00-\u9fff]。

标点符号

看来是需要抽半天空闲时间，仔细研究研究编码问题了。

参考

• 维基百科，Unicode 字符平面映射
• 维基百科，中日韓統一表意文字
• unicode.org，Unicode Technical Standard #18 - UNICODE REGULAR EXPRESSIONS
• unicode.org，Unihan Database Lookup
• Python 官方文档，7.8.3. Standard Encodings
• Python 官方文档，7.1.3.1. Format Specification Mini-Language
• CSDN 论坛，汉字的 unicode 码范围是多少

#16 Python 单例模式

使用全局变量

我个人比较喜欢的方式，简单高效。

class Application:
    pass

APP = Application()

装饰器（使用特定方法来获取对象实例）

import threading

class singleton:
    _instance_lock = threading.Lock()

    def __init__(self, decorated):
        self._decorated = decorated

    def instance(self):
        if not hasattr(self, '_instance'):
            with singleton._instance_lock:
                if not hasattr(self, '_instance'):
                    self._instance = self._decorated()
        return self._instance

    def __call__(self):
        raise TypeError('Singletons must be accessed through `instance()`.')

    def __instancecheck__(self, inst):
        return isinstance(inst, self._decorated)

    def __getattr__(self, name):
        if name == '_instance':
            raise AttributeError("'singleton' object has no attribute '_instance'")
        return getattr(self.instance(), name)

@singleton
class Application:
    pass

Application.instance()

装饰器

def singleton(cls):
    _instances = {}

    def _inner(*args, **kargs):
        if cls not in _instances:
            _instances[cls] = cls(*args, **kargs)
        return _instances[cls]

    return _inner

元类

class Application:
    _instance_lock = threading.Lock()

    def __new__(cls, *args, **kwargs):
        if not hasattr(cls, "_instance"):
            with cls._instance_lock:
                if not hasattr(cls, "_instance"):
                    cls._instance = object.__new__(cls)
        return cls._instance

改成通用方式：

import threading

class SingletonMeta(type):
    _instances = {}
    _instance_lock = threading.Lock()

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            with cls._instance_lock:
                if cls not in cls._instances:
                    instance = super().__call__(*args, **kwargs)
                    cls._instances[cls] = instance
        return cls._instances[cls]

class Application(metaclass=SingletonMeta):
    pass

#15 SQLAlchemy LIKE

例子：搜索用户表 user 中字段 phone 包含 520 的行。

SQL

和 pymysql 等库一样的用：

keyword = '520'
conn.execute('select * from user where phone like "%%%s%%";' % keyword)
conn.execute('select * from user where phone like "%%%s%%";', keyword)

.like 方法

q = session.query(model.User.id, model.User.phone).filter(model.User.phone.like(f'%{keyword}%'))
qs = qs.all()
# print(qs.statement)
# SELECT "user".id, "user".phone
# FROM "user"
# WHERE "user".phone LIKE :phone_1

对应的大小写不敏感方法有 ilike (lower("user".phone) LIKE lower(:phone_1))
还有：not_like, not_ilike

.contains 方法

print(session.query(model.User.id, model.User.phone).filter(model.User.phone.contains(keyword)).statement)
# SELECT "user".id, "user".phone
# FROM "user"
# WHERE ("user".phone LIKE '%' || :phone_1 || '%')

.regexp_match 方法（1.4 新增）

对应的是 MySQL 支持的 REGEXP 操作符。

print(session.query(model.User.id, model.User.phone).filter(model.User.phone.regexp_match(keyword)).statement)
session.query(model.User.id, model.User.phone).filter(model.User.phone.regexp_match(keyword)).all()
# SELECT "user".id, "user".phone
# FROM "user"
# WHERE "user".phone <regexp> :phone_1

.startswith 和 .endswith

print(session.query(model.User.id, model.User.phone).filter(model.User.phone.startswith(keyword)).statement)
# SELECT "user".id, "user".phone
# FROM "user"
# WHERE ("user".phone LIKE :phone_1 || '%')
print(session.query(model.User.id, model.User.phone).filter(model.User.phone.endswith(keyword)).statement)
# SELECT "user".id, "user".phone
# FROM "user"
# WHERE ("user".phone LIKE '%' || :phone_1)

.match 方法

对应的是数据库的 MATCH (col1,col2,...) AGAINST (expr [search_modifier]) 全文索引方法。
对单字段同样可用，不过需要先建立 FULLTEXT 索引。

print(session.query(model.User.id, model.User.phone).filter(model.User.phone.match(keyword)).statement)
session.query(model.User.id, model.User.phone).filter(model.User.phone.match(keyword)).all()
# SELECT "user".id, "user".phone
# FROM "user"
# WHERE "user".phone MATCH :phone_1

参考资料与拓展阅读

• https://docs.sqlalchemy.org/en/14/core/sqlelement.html
• https://dev.mysql.com/doc/refman/8.0/en/regexp.html

#14 SQLAlchemy

数据库连接：Engine

创建 engine 相当于通过适配层对接了原生数据库接口。当 excute 方法和 connect 方法首次调用时，Engine 建立了一个正式的 DBAPI 连接到数据库，之后的所有数据库交互都是通过这个连接发出。
使用 ORM 时，Engine 是个幕后工作者，也就是说一旦创建之后，我们一般不会再直接接触到 Engine。

from sqlalchemy import create_engine
engine = create_engine('sqlite:///:memory:', echo=True)  # echo => logging

连接字符串（DB URL）

会话（Session）

http://sqlalchemy.readthedocs.org/en/latest/orm/session_basics.html

Session 是一个基于指定数据库连接的一个工作区。文档上有这么一个例子：如果将一个应用程序线程当作一场派对中的一个来宾，会话就是这个来宾手上的盘子，数据就是盘子中的食物，数据库就是厨房。

何时创建，何时提交，何时关闭？

Session = sessionmaker(bind=engine)
# 等同于：
# Session = sessionmaker()
# Session.configure(bind=engine)
session = Session()

定义模型（数据库映射）

基类

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

模型的定义

from sqlalchemy import Column, Integer, String


class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    fullname = Column(String)
    password = Column(String)

    def __repr__(self):
        return "<User(name='%s', fullna

表结构（Schema）

又出来一个幕后工作者：Mapper，通过 模型类.__table__ 访问。

print type(User.__table__)  # <class 'sqlalchemy.sql.schema.Table'>
print repr(User.__table__)
# Table('users', MetaData(bind=None),
#       Column('id', Integer(), table=<users>, primary_key=True, nullable=False),
#       Column('name', String(), table=<users>),
#       Column('fullname', String(), table=<users>),
#       Column('password', String(), table=<users>), schema=None)

数据库操作

1. 添加

u = User(name='ed', fullname='Ed Jones', password='password')
session.add(u)

# 一次添加多条记录
session.add_all([
    User(name='wendy', fullname='Wendy Williams', password='foobar'),
    User(name='mary', fullname='Mary Contrary', password='p@55vv0rd'),
    User(name='fred', fullname='Fred Green', password='123456')
])
print session.new
print session.query(User).all()

2. 查找

2.1 查询对象 Query

for name, fullname in session.query(User.name, User.fullname):
    print name.ljust(10), fullname

for row in session.query(User, User.name):
    print row.User, row.name

直接遍历 Query 对象就相当于按照默认顺序获取所有对象。

2.2 条件：filte_by

qs = session.query(User).filter_by(name='ed')
users = qs.all()
our_user = qs.first()

2.3 更加灵活的 filter

允许使用 Python 表达式和类属性！

query = session.query(User.name)
queryset = query.filter(User.fullname == 'Ed Jones')
for name, in queryset:
    print name

通用 Filter 操作符

• Equal
  query.filter(User.name == 'ed')
query.filter(User.name != 'ed')
• LIKE
  query.filter(User.name.like('%ed%'))
• IN / NOT IN
  query.filter(User.name.in_(['ed', 'wendy', 'jack']))
query.filter(User.name.in_(session.query(User.name).filter(User.name.like('%ed%'))))
query.filter(~User.name.in_(['ed', 'wendy', 'jack']))
• NULL
  query.filter(User.name == None)
query.filter(User.name.is_(None))
query.filter(User.name != None)
query.filter(User.name.isnot(None))
• AND
  query.filter(User.name == 'ed').filter(User.fullname == 'Ed Jones')
query.filter(and_(User.name == 'ed', User.fullname == 'Ed Jones')) # from sqlalchemy import and_
• OR
  query.filter(User.name == 'ed', User.fullname == 'Ed Jones')
query.filter(or_(User.name == 'ed', User.name == 'wendy')) # from sqlalchemy import or_
• MATCH:
  query.filter(User.name.match('wendy'))
  注意：match 使用了数据库相关的内容，可能在不同数据库后端上有不同的效果，而且部分数据库不支持。

2.4 排序：order_by

for instance in session.query(User).order_by(User.id):
    print instance.name.ljust(10), instance.fullname

2.5 别名

label = User.name.label('name_label')
for row in session.query(label).all():
    print(row.name_label)

from sqlalchemy.orm import aliased
user_alias = aliased(User, name='user_alias')
for row in session.query(user_alias, user_alias.name).all():
    print row.user_alias

2.6 LIMIT 和 OFFSET

和 Django 一样，使用切片实现。

2.7 其他方法

• .all() 返回查找到的所有纪录组成的列表。
• .first() 返回查找到的第一条记录，没有找到返回 None。
• .one() 只查找到一条记录时，返回该条记录。否则抛出异常：
  sqlalchemy.orm.exc.NoResultFound: No row was found for one()
sqlalchemy.orm.exc.MultipleResultsFound: Multiple rows were found for one()
• .scalar() 用 .one() 方法模拟出 .first() 方法的效果。不同的是，.first() 方法只获取那一条（LIMIT 1），而调用 .one() 方法却获取了所有记录，不过只返回了第一条。这是不一样的！

2.8 对原生 SQL 的支持

• text
• from_statement
• params 为 Query 增加参数

from sqlalchemy import text

# 例 1，简单用法
for user in session.query(User).filter(text("id<224")).order_by(text("id")).all():
    print user.name

# 例 2，定义参数
session.query(User).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(User.id).one()

# 例 3，此时 Query 几乎只检测字段是否合法～
sql = text("SELECT * FROM users where name=:name")
session.query(User).from_statement(sql).params(name='ed').all()
session.query("id", "name").from_statement(sql).params(name='ed').all()

3. 更新

u = session.query(User).filter_by(name='fred')[0]
u.password = '12345678'
session.commit()
print u in session
u = session.query(User).filter_by(name='fred')[0]
print u.password

4. 删除

jack = session.query(User).filter_by(name='jack').one()
session.delete(jack)
session.query(User).filter_by(name='jack').count()

5. Count

session.query(User).filter(User.name.like('%ed')).count()
from sqlalchemy import func
session.query(func.count(User.name), User.name).group_by(User.name).all()
session.query(func.count('*')).select_from(User).scalar()
session.query(func.count(User.id)).scalar()

6. 事务：commit 和 rollback

commit

u = session.query(User).all()[-1]
u.password = '123456'
print session.dirty  # IdentitySet
# 此时 add_all 的新增操作实际上还没有提交
# session.commit()  # 提交到数据库

rollback

u = session.query(User).all()[-1]
u.password = '123456'
session.rollback()  # 撤销更新
print session.dirty
print 'After Rollback, PWD:'
print u.password  # 会先查找一遍～

关系

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship, backref

class Address(Base):
    __tablename__ = 'addresses'

    id = Column(Integer, primary_key=True)
    email_address = Column(String, nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))

    user = relationship("User", backref=backref('addresses', order_by=id))

    def __repr__(self):
        return "<Address(email_address='%s')>" % self.email_address

# 或者到 User 中声明，如下：
# class User(Base):
#     addresses = relationship("Address", order_by="Address.id", backref="user")

关联操作

jack = User(name='jack', fullname='Jack Bean', password='qwerty')
print jack.addresses
jack.addresses = [
    Address(email_address='jack@google.com'),
    Address(email_address='j25@yahoo.com'),
]
print jack.addresses[1]
print jack.addresses[1].user
session.add(jack)
print '-' * 70
session.commit()  # 也会插入 Address 信息
print '-' * 70
jack = session.query(User).filter_by(name='jack').one()
print jack
print '-' * 70
print jack.addresses   # lazy loading

JOIN

for u, a in session.query(User, Address).\
        filter(User.id==Address.user_id).\
        filter(Address.email_address=='jack@google.com').\
        all():
    print u
    print a
# <User(name='jack', fullname='Jack Bean', password='gjffdd')>
# <Address(email_address='jack@google.com')>

Join 也可以如下表示：session.query(User).join(Address)

如果两张表没有声明的关联，或者有多个关联，最好使用下面的形式：

query.join(Address, User.id==Address.user_id)  # explicit condition
query.join(User.addresses)  # specify relationship from left to right
query.join(Address, User.addresses)  # same, with explicit target
query.join('addresses')  # same, using a string

默认是 INNER JOIN：

session.query(User).join(Address). \
    filter(Address.email_address == 'jack@google.com'). \
    all()
# SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password
# FROM users
# JOIN addresses ON users.id = addresses.user_id
# WHERE addresses.email_address = 'jack@google.com'

使用其他连接方式：

query.outerjoin(User.addresses)   # LEFT OUTER JOIN

或者使用 join 方法的其他参数，具体信息，在后面章节会涉及。

aliased 又出现了

from sqlalchemy.orm import aliased
adalias1 = aliased(Address)
adalias2 = aliased(Address)
for username, email1, email2 in \
    session.query(User.name, adalias1.email_address, adalias2.email_address).\
    join(adalias1, User.addresses).\
    join(adalias2, User.addresses).\
    filter(adalias1.email_address=='jack@google.com').\
    filter(adalias2.email_address=='j25@yahoo.com'):
    print username, email1, email2
# jack jack@google.com j25@yahoo.com

子查询

from sqlalchemy.sql import func
stmt = session.query(Address.user_id, func.count('*').label('address_count')).\
    group_by(Address.user_id).subquery()
for u, count in session.query(User, stmt.c.address_count).\
        outerjoin(stmt, User.id==stmt.c.user_id).\
        order_by(User.id):
    print u, count

使用子查询

stmt = session.query(Address).\
    filter(Address.email_address != 'j25@yahoo.com').subquery()
adalias = aliased(Address, stmt)
for user, address in session.query(User, adalias).\
        join(adalias, User.addresses):
    print user
    print address

EXIST

from sqlalchemy.sql import exists
stmt = exists().where(Address.user_id==User.id)
for name, in session.query(User.name).filter(stmt):
    print name

for name, in session.query(User.name).filter(User.addresses.any()):
    print name

for name, in session.query(User.name).\
    filter(User.addresses.any(Address.email_address.like('%google%'))):
    print name

session.query(Address).filter(~Address.user.has(User.name=='jack')).all()

其他方法

query.filter(User.addresses.contains(someaddress))
query.filter(User.addresses.any(Address.email_address == 'bar'))
query.filter(User.addresses.any(email_address='bar'))
query.filter(Address.user.has(name='ed'))
session.query(Address).with_parent(someuser, 'addresses')

预加载

Eager Loading 直译过来，表示立即载入。

http://sqlalchemy.readthedocs.org/en/latest/orm/loading_relationships.html

默认是 Lazy Loading。

1. subqueryload

from sqlalchemy.orm import subqueryload
jack = session.query(User).\
    options(subqueryload(User.addresses)).\
    filter_by(name='jack').one()

subqueryload() when used in conjunction with limiting such as Query.first(), Query.limit() or Query.offset() should also include Query.order_by() on a unique column in order to ensure correct results.

2. joinedload

默认 LEFT OUTER JOIN。

from sqlalchemy.orm import joinedload
jack = session.query(User).\
    options(joinedload(User.addresses)).\
    filter_by(name='jack').one()

joinedload() is not a replacement for join()
The join created by joinedload() is anonymously aliased such that it does not affect the query results. An Query.order_by() or Query.filter() call cannot reference these aliased tables - so-called “user space” joins are constructed using Query.join(). The rationale for this is that joinedload() is only applied in order to affect how related objects or collections are loaded as an optimizing detail - it can be added or removed with no impact on actual results. See the section The Zen of Eager Loading for a detailed description of how this is used.

3. Join + Eagerload

from sqlalchemy.orm import contains_eager
jacks_addresses = session.query(Address).\
    join(Address.user).\
    filter(User.name=='jack').\
    options(contains_eager(Address.user)).\
    all()

关联对象的删除

回到曾经的示例——删除 Jack 上来，如果 Jack 有几个地址，实际上 Jack 删除之后那几个地址的用户主键会被设置成 NULL，但是并没有删除。

#13 logging 不输出中文日志的问题

一个小问题。