首页 编程教程正文

python实用代码片段收集贴

piaodoo 编程教程 2020-02-02 12:25:31 795 0 python教程

这篇文章主要介绍了python实用代码片段收集贴,本文收集了如获取一个类的所有子类、计算运行时间、SQLAlchemy简单使用、实现类似Java或C中的枚举等实用功能代码,需要的朋友可以参考下

获取一个类的所有子类

复制代码 代码如下:

def itersubclasses(cls, _seen=None):
    """Generator over all subclasses of a given class in depth first order."""
    if not isinstance(cls, type):
        raise TypeError(_('itersubclasses must be called with '
                          'new-style classes, not %.100r') % cls)
    _seen = _seen or set()
    try:
        subs = cls.__subclasses__()
    except TypeError:   # fails only when cls is type
        subs = cls.__subclasses__(cls)
    for sub in subs:
        if sub not in _seen:
            _seen.add(sub)
            yield sub
            for sub in itersubclasses(sub, _seen):
                yield sub

简单的线程配合

复制代码 代码如下:

import threading
is_done = threading.Event()
consumer = threading.Thread(
    target=self.consume_results,
    args=(key, self.task, runner.result_queue, is_done))
consumer.start()
self.duration = runner.run(
        name, kw.get("context", {}), kw.get("args", {}))
is_done.set()
consumer.join() #主线程堵塞,直到consumer运行结束

多说一点,threading.Event()也可以被替换为threading.Condition(),condition有notify(), wait(), notifyAll()。解释如下:
复制代码 代码如下:

The wait() method releases the lock, and then blocks until it is awakened by a notify() or notifyAll() call for the same condition variable in another thread. Once awakened, it re-acquires the lock and returns. It is also possible to specify a timeout.
The notify() method wakes up one of the threads waiting for the condition variable, if any are waiting. The notifyAll() method wakes up all threads waiting for the condition variable.
Note: the notify() and notifyAll() methods don't release the lock; this means that the thread or threads awakened will not return from their wait() call immediately, but only when the thread that called notify() or notifyAll() finally relinquishes ownership of the lock.

复制代码 代码如下:

# Consume one item
cv.acquire()
while not an_item_is_available():
    cv.wait()
get_an_available_item()
cv.release()
# Produce one item
cv.acquire()
make_an_item_available()
cv.notify()
cv.release()

计算运行时间

复制代码 代码如下:

class Timer(object):
    def __enter__(self):
        self.error = None
        self.start = time.time()
        return self
    def __exit__(self, type, value, tb):
        self.finish = time.time()
        if type:
            self.error = (type, value, tb)
    def duration(self):
        return self.finish - self.start
with Timer() as timer:
    func()
return timer.duration()

元类

__new__()方法接收到的参数依次是:
当前准备创建的类的对象;
类的名字;
类继承的父类集合;
类的方法集合;

复制代码 代码如下:

class ModelMetaclass(type):
    def __new__(cls, name, bases, attrs):
        if name=='Model':
            return type.__new__(cls, name, bases, attrs)
        mappings = dict()
        for k, v in attrs.iteritems():
            if isinstance(v, Field):
                print('Found mapping: %s==>%s' % (k, v))
                mappings[k] = v
        for k in mappings.iterkeys():
            attrs.pop(k)
        attrs['__table__'] = name # 假设表名和类名一致
        attrs['__mappings__'] = mappings # 保存属性和列的映射关系
        return type.__new__(cls, name, bases, attrs)
class Model(dict):
    __metaclass__ = ModelMetaclass
    def __init__(self, **kw):
        super(Model, self).__init__(**kw)
    def __getattr__(self, key):
        try:
            return self[key]
        except KeyError:
            raise AttributeError(r"'Model' object has no attribute '%s'" % key)
    def __setattr__(self, key, value):
        self[key] = value
    def save(self):
        fields = []
        params = []
        args = []
        for k, v in self.__mappings__.iteritems():
            fields.append(v.name)
            params.append('?')
            args.append(getattr(self, k, None))
        sql = 'insert into %s (%s) values (%s)' % (self.__table__, ','.join(fields), ','.join(params))
        print('SQL: %s' % sql)
        print('ARGS: %s' % str(args))
class Field(object):
    def __init__(self, name, column_type):
        self.name = name
        self.column_type = column_type
    def __str__(self):
        return '<%s:%s>' % (self.__class__.__name__, self.name)
class StringField(Field):
    def __init__(self, name):
        super(StringField, self).__init__(name, 'varchar(100)')
class IntegerField(Field):
    def __init__(self, name):
        super(IntegerField, self).__init__(name, 'bigint')
class User(Model):
    # 定义类的属性到列的映射:
    id = IntegerField('id')
    name = StringField('username')
    email = StringField('email')
    password = StringField('password')
# 创建一个实例:
u = User(id=12345, name='Michael', email='test@orm.org', password='my-pwd')
# 保存到数据库:
u.save()

输出如下:

复制代码 代码如下:

Found model: User
Found mapping: email ==> <StringField:email>
Found mapping: password ==> <StringField:password>
Found mapping: id ==> <IntegerField:uid>
Found mapping: name ==> <StringField:username>
SQL: insert into User (password,email,username,uid) values (?,?,?,?)
ARGS: ['my-pwd', 'test@orm.org', 'Michael', 12345]

SQLAlchemy简单使用

复制代码 代码如下:

# 导入:
from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
# 创建对象的基类:
Base = declarative_base()
# 定义User对象:
class User(Base):
    # 表的名字:
    __tablename__ = 'user'
    # 表的结构:
    id = Column(String(20), primary_key=True)
    name = Column(String(20))
# 初始化数据库连接:
engine = create_engine('mysql+mysqlconnector://root:password@localhost:3306/test') # '数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名'
# 创建DBSession类型:
DBSession = sessionmaker(bind=engine)
# 创建新User对象:
new_user = User(id='5', name='Bob')
# 添加到session:
session.add(new_user)
# 提交即保存到数据库:
session.commit()
# 创建Query查询,filter是where条件,最后调用one()返回唯一行,如果调用all()则返回所有行:
user = session.query(User).filter(User.id=='5').one()
# 关闭session:
session.close()

WSGI简单使用和Web框架Flask的简单使用

复制代码 代码如下:

from wsgiref.simple_server import make_server
def application(environ, start_response):
    start_response('200 OK', [('Content-Type', 'text/html')])
    return '<h1>Hello, web!</h1>'
# 创建一个服务器,IP地址为空,端口是8000,处理函数是application:
httpd = make_server('', 8000, application)
print "Serving HTTP on port 8000..."
# 开始监听HTTP请求:
httpd.serve_forever()

了解了WSGI框架,我们发现:其实一个Web App,就是写一个WSGI的处理函数,针对每个HTTP请求进行响应。
但是如何处理HTTP请求不是问题,问题是如何处理100个不同的URL。
一个最简单和最土的想法是从environ变量里取出HTTP请求的信息,然后逐个判断。

复制代码 代码如下:

from flask import Flask
from flask import request
app = Flask(__name__)
@app.route('/', methods=['GET', 'POST'])
def home():
    return '<h1>Home</h1>'
@app.route('/signin', methods=['GET'])
def signin_form():
    return '''<form action="/signin" method="post">
              <p><input name="username"></p>
              <p><input name="password" type="password"></p>
              <p><button type="submit">Sign In</button></p>
              </form>'''
@app.route('/signin', methods=['POST'])
def signin():
    # 需要从request对象读取表单内容:
    if request.form['username']=='admin' and request.form['password']=='password':
        return '<h3>Hello, admin!</h3>'
    return '<h3>Bad username or password.</h3>'
if __name__ == '__main__':
    app.run()

格式化显示json

复制代码 代码如下:

print(json.dumps(data, indent=4))
# 或者
import pprint
pprint.pprint(data)

实现类似Java或C中的枚举

复制代码 代码如下:

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import itertools
import sys
class ImmutableMixin(object):
    _inited = False
    def __init__(self):
        self._inited = True
    def __setattr__(self, key, value):
        if self._inited:
            raise Exception("unsupported action")
        super(ImmutableMixin, self).__setattr__(key, value)
class EnumMixin(object):
    def __iter__(self):
        for k, v in itertools.imap(lambda x: (x, getattr(self, x)), dir(self)):
            if not k.startswith('_'):
                yield v
class _RunnerType(ImmutableMixin, EnumMixin):
    SERIAL = "serial"
    CONSTANT = "constant"
    CONSTANT_FOR_DURATION = "constant_for_duration"
    RPS = "rps"
if __name__=="__main__":
    print _RunnerType.CONSTANT

创建文件时指定权限

复制代码 代码如下:

import os
def write_to_file(path, contents, umask=None):
    """Write the given contents to a file
    :param path: Destination file
    :param contents: Desired contents of the file
    :param umask: Umask to set when creating this file (will be reset)
    """
    if umask:
        saved_umask = os.umask(umask)
    try:
        with open(path, 'w') as f:
            f.write(contents)
    finally:
        if umask:
            os.umask(saved_umask)
if __name__ == '__main__':
    write_to_file('/home/kong/tmp', 'test', 31)
    # Then you will see a file is created with permission 640.
    # Warning: If the file already exists, its permission will not be changed.
    # Note:For file, default all permission is 666, and 777 for directory.

多进程并发执行

复制代码 代码如下:

import multiprocessing
import time
import os
def run(flag):
    print "flag: %s, sleep 2s in run" % flag
    time.sleep(2)
    print "%s exist" % flag
    return flag
if __name__ == '__main__':
    pool = multiprocessing.Pool(3)
    iter_result = pool.imap(run, xrange(6))
    print "sleep 5s\n\n"
    time.sleep(5)
    for i in range(6):
        try:
            result = iter_result.next(600)
        except multiprocessing.TimeoutError as e:
            raise
        print result
    pool.close()
    pool.join()

运行时自动填充函数参数

复制代码 代码如下:

import decorator
def default_from_global(arg_name, env_name):
    def default_from_global(f, *args, **kwargs):
        id_arg_index = f.func_code.co_varnames.index(arg_name)
        args = list(args)
        if args[id_arg_index] is None:
            args[id_arg_index] = get_global(env_name)
            if not args[id_arg_index]:
                print("Missing argument: --%(arg_name)s" % {"arg_name": arg_name})
                return(1)
        return f(*args, **kwargs)
    return decorator.decorator(default_from_global)
# 如下是一个装饰器,可以用在需要自动填充参数的函数上。功能是:
# 如果没有传递函数的deploy_id参数,那么就从环境变量中获取(调用自定义的get_global函数)
with_default_deploy_id = default_from_global('deploy_id', ENV_DEPLOYMENT)   

嵌套装饰器

validator函数装饰func1,func1使用时接收参数(*arg, **kwargs),而func1又装饰func2(其实就是Rally中的scenario函数),给func2增加validators属性,是一个函数的列表,函数的接收参数config, clients, task。这些函数最终调用func1,传入参数(config, clients, task, *args, **kwargs),所以func1定义时参数是(config, clients, task, *arg, **kwargs) 
最终实现的效果是,func2有很多装饰器,每个都会接收自己的参数,做一些校验工作。

复制代码 代码如下:

def validator(fn):
    """Decorator that constructs a scenario validator from given function.
    Decorated function should return ValidationResult on error.
    :param fn: function that performs validation
    :returns: rally scenario validator
    """
    def wrap_given(*args, **kwargs):
        """Dynamic validation decorator for scenario.
        :param args: the arguments of the decorator of the benchmark scenario
        ex. @my_decorator("arg1"), then args = ('arg1',)
        :param kwargs: the keyword arguments of the decorator of the scenario
        ex. @my_decorator(kwarg1="kwarg1"), then kwargs = {"kwarg1": "kwarg1"}
        """
        def wrap_validator(config, clients, task):
            return (fn(config, clients, task, *args, **kwargs) or
                    ValidationResult())
        def wrap_scenario(scenario):
            wrap_validator.permission = getattr(fn, "permission",
                                                consts.EndpointPermission.USER)
            if not hasattr(scenario, "validators"):
                scenario.validators = []
            scenario.validators.append(wrap_validator)
            return scenario
        return wrap_scenario
    return wrap_given

inspect库的一些常见用法

inspect.getargspec(func) 获取函数参数的名称和默认值,返回一个四元组(args, varargs, keywords, defaults),其中:
args是参数名称的列表;
varargs和keywords是*号和**号的变量名称;
defaults是参数默认值的列表;

inspect.getcallargs(func[, *args][, **kwds]) 绑定函数参数。返回绑定后函数的入参字典。

python中的私有属性和函数

Python把以两个或以上下划线字符开头且没有以两个或以上下划线结尾的变量当作私有变量。私有变量会在代码生成之前被转换为长格式(变为公有),这个过程叫"Private name mangling",如类A里的__private标识符将被转换为_A__private,但当类名全部以下划线命名的时候,Python就不再执行轧压。而且,虽然叫私有变量,仍然有可能被访问或修改(使用_classname__membername),所以, 总结如下:

无论是单下划线还是双下划线开头的成员,都是希望外部程序开发者不要直接使用这些成员变量和这些成员函数,只是双下划线从语法上能够更直接的避免错误的使用,但是如果按照_类名__成员名则依然可以访问到。单下划线的在动态调试时可能会方便一些,只要项目组的人都遵守下划线开头的成员不直接使用,那使用单下划线或许会更好。

版权声明:

本站所有资源均为站长或网友整理自互联网或站长购买自互联网,站长无法分辨资源版权出自何处,所以不承担任何版权以及其他问题带来的法律责任,如有侵权或者其他问题请联系站长删除!站长QQ754403226 谢谢。

有关影视版权:本站只供百度云网盘资源,版权均属于影片公司所有,请在下载后24小时删除,切勿用于商业用途。本站所有资源信息均从互联网搜索而来,本站不对显示的内容承担责任,如您认为本站页面信息侵犯了您的权益,请附上版权证明邮件告知【754403226@qq.com】,在收到邮件后72小时内删除。本文链接:https://www.piaodoo.com/1498.html

评论

搜索

游戏网站源码,织梦网站源码,wordpress,wordpress主题,wordpress下载,wordpress插件,wordpress.com,wordpress模板,wordpress教程,wordpress 主题,wordpress安装,wordpress 模板,wordpress 插件,wordpress主题下载,wordpress企业主题,wordpress seo,wordpress主题开发,wordpress theme,wordpress论坛,wordpress 企业主题,wordpress主机,wordpress中文主题,wordpress cms主题,wordpress plugin,wordpress 主题下载,wordpress 主机,wordpress空间,wordpress mu,wordpress 模版,wordpress汉化主题,wordpress淘宝客主题,wordpress 空间,wordpress代码,WORDPRESS HOSTING,wordpress优点,wordpress安卓客户端,wordpress技巧,wordpress换空间,wordpress themes,网站模板,ppt模板网站,模板网站,企业网站模板,网站设计模板,免费网站模板,个人网站模板,ppt模板下载网站,网站模板下载,公司网站模板,门户网站模板,学校网站模板,网站首页模板,网站模板免费下载,旅游网站模板,网站后台模板,免费网站模板下载,传奇网站模板,网站建设模板,外贸网站模板,网站 模板,个人主页网站模板,个人网站模板下载,政府网站模板,音乐网站模板,导航网站模板,免费企业网站模板,企业网站模板下载,手表网站模板,韩国网站模板,汽车网站模板,教育网站模板,网站后台管理模板,班级网站模板,新闻网站模板,房产中介网站模板,旅游网站模板下载,工艺品网站模板,电子商务网站模板,旅游网站设计模板,团购网站模板,flash网站模板,个人网站设计模板,婚庆网站模板,广告公司网站模板,商业网站模板,手机网站模板,免费模板网站推荐,ppt免费模板网站推荐,织梦网站模板,html网站模板建站,网站html模板,免费个人网站模板,公司网站源码,sns源码,彩票网站源码,周易网站源码,源码基地,交友源码,学校网站源码,asp.net 源码,源码天下,jsp网站源码,论坛源码下载,广告联盟源码,建站源码,delphi源码,源码爱好者,酷源码,net源码,源码超市,医疗网站源码,flash源码,搜源码,源码程序,dede源码,新闻网站源码,易语言源码大全,旅游网站源码下载,flash 源码,免费源码论坛,android游戏源码,电脑维修网站源码,30源码网,股票软件源码,卖源码,源码教程,安居客 源码,vip源码,家教源码,.net源码下载,Web源码,网络公司源码,佛教网站源码,android源码学习,房产源码,钓鱼网站源码,775源码屋,web游戏源码,成品网站 源码78w78不用下载,h5游戏网站源码,asp网站源码下载,webgame源码,电子商务网站源码,vb.net源码,乐嘿源码,8a商业源码论坛,fbreader源码,在线客服系统 源码,google源码,.net网站源码,快递查询源码,源码搜藏网,dede整站源码,周易 源码,52源码论坛,财经网站源码,织梦下载站源码,qq钓鱼网站源码,flash游戏源码,房产网源码,源码搜搜,电子商务源码,团购网站源码,团购网源码,jsp源码下载,jsp源码,h站源码,8a源码,婚纱摄影网站源码,易语言盗号源码,x站源码,qq空间psd源码,免费商业源码,笑话网站源码,源码集合,源码家园,啊哦源码,星期六源码,源码熊,阿奇源码,百分百源码网,一手日源码资源,旅行网站源码,b站工程源码泄露,新站长源码,8a商业源码,asp论坛源码,flash源码下载,404源码社区,创业网站源码,php网页源码,易支付源码,成品网站w灬源码,免费CMS成品网站源码,成品网站W灬源码1688仙踪林,成品APP短视频源码下载网站,成品网站源码1688可靠吗,免费B2B网站源码,成品APP直播源码下载,国外儿童网站源码在线,成品网站W灬源码1688,源码,成品网站w灬 源码1688,免费源码网站都有哪些,成品网站源码78W78隐藏通道1,网站源码,源码网,源码网站,源码时代,源码之家,源码下载,php源码,易语言源码,源码论坛,源码是什么,商城源码,论坛源码,源码交易,源码站,源码库,免费源码,免费网站ja**源码大全,ja**源码,成品网站w灬源码1377,a5源码,站长源码,成品网站源码78W78隐藏通道1APP,源码分享,网站源码下载,源码中国,asp源码,源码社区,企业网站源码,php源码下载,成品app直播源码搭建,在线观看视频网站源码2021,旅游网站源码,安卓源码,通达信选股公式源码,神马影院php源码,c#源码,成品网站w灬源码1688网页,php 源码,网页游戏源码,android源码下载,源码吧,视频源码大全,成品短视频APP源码搭建,asp源码下载,私服源码,电脑维修源码,个人主页源码,源码出售,php网站源码,刀客源码,网址导航源码,导航网站源码,源码天空,asp 源码,软件源码,精品源码,成品网站源码1688自动跳转,个人网站源码,源码哥,在线考试系统源码,cms源码,c# 源码,商业源码,vb源码,门户网站源码,音乐网站源码,中国源码,安卓源码下载,asp网站源码,在线客服源码,电影网站源码,免费源码下载,整站源码,源码交易网,易语言源码网,.net源码,在线客服系统源码,淘客源码,卡盟源码,网站源码出售,vb源码下载,莎莎源码,熊猫烧香源码,asp.net源码,商业源码网,外贸网站源码,61源码网,zblog模板,zblog企业模板,帝国cms模板,帝国cms插件,discuz模板