一些Python必學知識點，送給正在學習Python中的你

面向物件

#!/usr/bin/env python
# -*- coding: utf-8 -*-
from _ast import Num

class Student(object): # object 是所有物件的基類/python學習群125240963
    def __init__(self, name, score, id): # 建構函式
        self.name = name
        self.score = score
        self.__id = id # 成員變數前加 __ 表示私有變數，無法從外部訪問

    def print_score(self):
        print '%s, %s' % (self.name, self.score)
    @property
    def id(self):
        return self.__id
    @id.setter
    def id(self, id):
        self.id = id
        
bart = Student('Bart', 95.0, 1) # 例項化類
list = Student('List', 87.1, 2)
bart.print_score() # 成員函式呼叫
list.print_score()
bart.sex = '女' # 可以動態的新增屬性，但是隻對當前物件有效
print bart.sex
# print list.sex # 丟擲異常AttributeError

print bart.name
#print bart.__id # 丟擲異常AttributeError
print bart.id

class Animal(object):
    def run(self):
        print 'Animal is running...'

class Dog(Animal):
    def run(self): # 覆蓋父類方法
        print 'Dog is running...'

    def __len__(self): # 要獲取自定義型別的len方法  需要自己編寫 __len__()函式
        return 3

class Cat(Animal):
    def run(self):
        print 'Cat is running...'

class Tortoise(Animal):
    def run(self):
        print 'Tortoise is running...'

dog = Dog()
cat = Cat()
dog.run()
cat.run()

def run_(animal):
    animal.run()

run_(Dog())
run_(Cat())
run_(Tortoise())


# 獲取物件資訊
print '---獲取物件資訊---'
print type(123)
print type('abc')
print type([])
print type(str)
print type(Animal)
print type(dog)

# 每一種內建型別都在 types 模組有定義
import types
print type(123) == types.IntType
print type(Animal) == types.TypeType

# 判斷型別 isinstance()
print '---isinatance()---'
print isinstance(dog, Animal)
print isinstance(dog, (Dog, Animal)) # 可以判斷是否是中的一種型別

# 獲取物件的所有屬性和方法
print '---dir()---'
print dir(dog)

print len(dog)

# getattr() setattr() hasattr()
try:
    print hasattr(bart, name)
except NameError:
    pass

# 可以動態的給物件繫結屬性
class A(object):
    pass
print '---動態繫結屬性---'
a = A()
b = A()
a.name = 'abc' # 僅對當前物件有效
print a.name
try:
    print b.name
except AttributeError:
    pass
print '---動態繫結方法---'
def set_age(self, age): # 定義方法體
    self.age = age
from types import MethodType # 匯入模組方法
a.set_age = MethodType(set_age, a, A) # 此方式只對當前例項有效
a.set_age(24)
print a.age

# 對所有例項都有效的動態繫結方法的方式 給類繫結
def set_score(self, score):
    self.score = score
A.set_score = MethodType(set_score, None, A)
b.set_score(54)
print b.score

# 限制動態建立屬性  __slots__ 變數 只對當前類有效，對子類無效
print '---限制---'
class B(object):
    __slots__ = ('name', 'age')
c = B()
c.name = 'ad'
c.age = 12
try:
    c.score = 'adf'
except AttributeError:
    print 'AttributeError'

# @property 裝飾器  將方法變為屬性，可通過直接屬性呼叫，並可檢測引數值
print '
 
[email protected]'
class C(object):
    def __init__(self):
        self.__score = 10
        
    def __len__(self):
        return 9
    
    @property # 只定義它，就只有getter
    def score(self):
        return self.__score
 
    @score.setter
    def score(self, value):
        if not isinstance(value, int):
            raise ValueError('score must be integer!')
        if value < 0 or value > 100:
            raise ValueError('score must be between 0 ~ 100')
        self.__score = value

cc = C()
cc.score = 60
print cc.score
try:
    cc.score = 9999
    print cc.score
except ValueError:
    print 'ValueError'

print len(cc)
    
# Python 支援多重繼承

# 類 內建屬性
print '類內建屬性'
# __repr__ 列印例項的地址，但是對於使用者而言無用，因此自定義__str__方法，讓__repr__ = __str__
class D(object):
    def __init__(self, name):
        self.name = name
        
    def __str__(self):
        return 'D object (name = %s)' % self.name
    __repr__ = __str__
    
print D('wangh') 
# 如果需要將類 作為可迭代的物件，需要實現__iter__(self) 及 next(self) 方法
class E(object):
    def __init__(self):
        self.a = 0
        self.b = 1
                
    def __iter__(self):
        return self
    
    def next(self):
        tmp = self.b
        self.b = self.a + self.b
        self.a = tmp
       # self.a, self.b = self.b, self.a + self.b # 先計算在加上 同上
        if self.a > 100000: 
            raise StopIteration()
        return self.a
    
    def __getitem__(self, n): # 使類可以使用索引取值， 但是需要對傳入的引數做型別判斷，如切片 isinstance(n, slice) 
        a, b = 1, 1
        for x in range(n):
            a, b = b, a + b
        return a
        # 還有負數等  與之對應的方法__setitem__   __delitem__ 

for n in E():
    print n

print E()[10]
# 對例項進行呼叫 需要實現 __call__ 方法
class F(object):
    def __init__(self, num):
        self.num = num
        
    def __call__(self):
        return 'Num = %d' % self.num
    
f = F(123)
print f()
# callable 函式  可以判斷 物件是否可以被呼叫
print callable(f)

# 使用 type()函式動態建立類
print '使用 type()函式動態建立類'
# 先定義函式
def fn(self, name = 'world'):
    print 'Hello %s' % name
Hello = type('Hello', (object,), dict(hello = fn)) # 建立Hello class
h = Hello()
h.hello()
# 要建立類 type() 需要傳入三個引數
# 1 class 的名稱 
# 2 繼承的父類集合，如果只有一個 記住  元祖的 ,
# 3 方法名和函式繫結
 

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# 錯誤處理機制 try  except  finally
print '-----錯誤處理-----'

try:
    r = 10 / 1
    print 'try...'
except ZeroDivisionError, e:
    print e
else: # else 語句塊 在沒有錯誤時執行
    print 'else...'
finally:
    print 'finally...'
print 'End'  

# 錯誤基類 BaseException

print '使用內建logging模組列印錯誤資訊'
import logging
logging.basicConfig(level=logging.INFO)

def foo(s):
    return 10 / int(s)

def bar(s):
    return foo(s) * 2

def main():
    try:
        bar('0')
    except StandardError, e:
        print e
        logging.exception(e) # 可以打印出堆疊資訊
main()
print 'End'

# 丟擲錯誤 raise 可以丟擲錯誤物件，也可以單獨使用 raise 用於將錯誤丟擲，供上層處理
def func1(s):
    return 10 / int(s)
def func2(s):
    try:
        func1(s)
    except ZeroDivisionError:
        print 'Error'
        raise

def main():
    try:
        func2('0')
    except ZeroDivisionError, e:
        logging.exception(e)
main()
print '---------------'

print '-----除錯-----'
# 使用 print
# 使用斷言， -O 引數關閉斷言
try:
    n = 0
    assert (n != 0) # 丟擲AssertError異常
except AssertionError, e:
    logging.exception(e)
else:
    10 / n
# 使用logging模組
print '---------------'
# import logging 
# logging.basicConfig(level=logging.INFO)
s = '0'
n = int(s)
logging.info('n = %d' % n)
print 10 / n
    
print '-------------'
 

正則表示式

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# 1 如果直接給出字元，則表示精確匹配
# 2 \d 表示數字， \w 表示字母或數字， . 可以匹配任意字元，\s 表示空白符，包括Tab
# 3 * 表示任意長度，+ 表示至少一個字元，？ 表示0或1個字元，{n} 表示n個字元，{n,m} 表示n到m個字元
# 如 \d{3}\s+\d{3,8}
# - 是特殊字元，需要\轉義 \-

# 更精確的匹配範圍 表示 []
# 如 [0-9a-zA-Z\_] 可以匹配一個數字，字母，或者下劃線
# A|B 可以匹配A，B
# ^ 表示行的開頭， ^\d 表示必須以數字開頭
# $ 表示行的結束，\d$ 表示必須以數字結束

# re 模組
# r作為字串的字首，可省略轉義
import re

print re.match(r'^\d{3}\-\d{3,8}$', '010-12345')
print re.match(r'^\d{3}\-\d{3,8}$', '010 12345')

# 切割字串功能比純字串更強
print 'a b   c'.split(' ')
print re.split(r'\s+', 'a b   c')
print re.split(r'[\s\,]+', 'a, b  c,, c')

#!/usr/bin/env python
# -*- coding: utf-8 -*-

type = 'MySQL'

if type == 'sqlite3':
    # 匯入驅動
    import sqlite3
    # 連線到資料庫 test.db 如果檔案不存在，則會自動建立
    conn = sqlite3.connect('test.db')
    # 建立一個cursor
    cursor = conn.cursor()
    # 執行建立表
    cursor.execute('create table user (id varchar(20) primary key, name varchar(20))')
    # 執行插入資料
    cursor.execute("insert into user (id, name) values ('1', 'wangh')")
    # 通過rowcount獲取影響的行數
    print cursor.rowcount
    # 關閉cursor
    cursor.close()
    # 提交事物
    conn.commit()
    # 關閉connection
    conn.close()
    
    conn1 = sqlite3.connect('test.db')
    cursor1 = conn1.cursor()
    # 執行查詢語句
    cursor1.execute("select * from user where id=?", '1')
    # 獲取結果集
    value = cursor1.fetchall()
    print value
    cursor1.close()
    conn1.close()

elif type == 'MySQL':
    import mysql.connector
    
    conn = mysql.connector.connect(user = 'root', password = 'wangh', database = 'test', use_unicode = True)
    cursor = conn.cursor()
    #cursor.execute("create table user (id varchar(20) primary key, name varchar(20))")
    # 插入記錄  MySQL的佔位符為 %s
    cursor.execute('insert into user(id, name) values(%s, %s)', ['1', 'kongl'])
    print cursor.rowcount
    conn.commit()
    cursor.close()
    
    # 查詢
    cursor = conn.cursor()
    cursor.execute('select * from user')# where id = %s' % '1')
    values = cursor.fetchall()
    print values
    cursor.close()
    conn.close()
    
    
else:
    pass

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# 簡單檔案及不重要資料 序列化和反序列化模組 cPickle 或 pickle
try:
    import cPickle as pickle
except ImportError:
    import pickle

d = dict(name = 'Bob', age = 20, score = 80)
print d
s = pickle.dumps(d)
print s
with open('dump.txt', 'w') as f:
    f.write(s)

with open('dump.txt', 'r') as fread:
    rs = fread.read()
rd = pickle.loads(rs)
print rd

with open('dump.txt', 'r') as fread:
    print pickle.load(fread)

# 重要資料 使用 JSON 模組 儲存，也適用於不同 語言的傳遞
print '---------------'
import json
d = dict(name = 'Wangh', age = 20, score = 80)
json_str = json.dumps(d)
print json_str

json_d = json.loads(json_str)
print json_d
# 注：反序列化得到的字串是unicode 需要轉換 為str

# 將類序列化需要 自己編寫預設序列化函式
class A(object):
    def __init__(self, name, age):
        self.name = name
        self.age = age
        
    def serial(self):
        return {
                'name' : self.name,
                'age' : self.age
                }
        
a = A('kongl', 24)
json_str = json.dumps(a, default=A.serial)
print json_str
print json.dumps(a, default = lambda obj : obj.__dict__) # 推薦此種方式，則不需要定義函式，但要排除 __slots__ 的class

def dict2obj(d):
    return A(d['name'], d['age'])

b = json.loads(json_str, object_hook = dict2obj)
print b.name
print b.age

多程序

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# os 模組封裝了 fock() 函式，但只適用於Unix/Mac 系統，且只能是Linux版的 Python 才有os.fork()函式

import os

if os.name == 'posix':
    print 'Process (%s) start...' % os.getpid()
    pid = os.fork()
    if pid == 0: # 子程序
        print "I'm child process (%s) and my parent is (%s)" % (os.getpid(), os.getppid())
    else:
        print 'I (%s) just create a child process (%s)' % (os.getpid(), pid)

# 跨平臺的 多程序模組 multiprocessing
from multiprocessing import Process
import os

#子程序要執行的程式碼
def run_proc(name):
    print 'Run child process %s (%s)...' % (name, os.getpid())

if __name__ == '__main__':
    print 'Parent process %s' % os.getpid()
    #建立程序
    p = Process(target = run_proc, args = ('test',))
    print 'Process will start'
    p.start()
    p.join() # 等待子程序的結束
    print 'Process end'
    
# 程序池
#程序池的大小，預設等於處理器的 核心數，可改變 p = pool(5)
from multiprocessing import Pool
import os, random, time

def long_time_task(name):
    print 'Run task %s (%s)' % (name, os.getpid())
    start = time.time()
    time.sleep(random.random() * 3)
    end = time.time()
    print 'Task %s runs %0.2f seconds' % (name, (end - start))
    
if __name__ == '__main__':
    print 'Parent process %s' % os.getpid()
    p = Pool(9)
    for i in range(5):
        p.apply_async(long_time_task, args = (i,))
    print 'Wait for all subprocess done...'
    p.close() # close 後進不能在加入process
    p.join()
    print 'All subprocess end...'
    
# 程序間通訊 Queue 或 Pipes
from multiprocessing import Process, Queue
import os, random, time

def write(q):
    for value in ['a', 'b', 'c']:
        print 'Put %s to queue...' % value
        q.put(value)
        time.sleep(4)

def read(q):
    while True:
        value = q.get(True)
        print 'Get %s from queue...' % value
        
if __name__ == '__main__':
    # 父程序建立queue，並傳給子程序
    q = Queue()
    pw = Process(target = write, args = (q,))
    pr = Process(target = read, args = (q,))
    # 啟動寫程序
    pw.start()
    # 啟動讀程序
    pr.start()
    # 等待結束
    pw.join()
    # 死迴圈強行結束
    pr.terminate()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# 低階模組 thread 高階模組 threading(常用)
import threading, time

# 執行緒執行程式碼
def loop():
    print 'thread %s is running...' % threading.current_thread().name
    n = 0
    while n < 5:
        n += 1
        print 'thread %s >>> %s' % (threading.current_thread().name, n)
        time.sleep(1)
    print 'thread %s end' % threading.current_thread().name
    
if __name__ == '__main__':
    print 'thread %s is running...' % threading.current_thread().name
    t = threading.Thread(target = loop, name = 'LoopThread')
    t.start()
    t.join()
    print 'thread %s end' % threading.current_thread().name

# 鎖 lock = threading.Lock()  lock.acquire()  lock.release

# ThreadLocal 解決不同執行緒處理同一個物件時  免去 傳參 及 全域性變數的方式、  常用於：每個執行緒繫結一個數據庫連線，HTTP請求，使用者身份資訊等
# 建立全域性ThreadLocal物件 
local_school = threading.local()

def process_student():
    print 'Hello %s (in %s)' % (local_school.student, threading.current_thread().name)

def process_thread(name):
    # 繫結TrheadLocal的student
    local_school.student = name
    process_student()

t1 = threading.Thread(target = process_thread, args = ('wangh',), name = 'ThreadA')
t2 = threading.Thread(target = process_thread, args = ('kongl',), name = 'ThreadB')
t1.start()
t2.start()
t1.join()
t2.join()

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import multiprocessing

# multiprocessing 的 子模組 managers 將網路通訊封裝，可以靜默升級 在同一臺機器上的 多程序程式，如 只需要將 Queue 通過managers暴露在網上即可

type = 'd'
if type == 'server':
    # 伺服器端，負責建立Queue，釋出到網路，並向其中寫任務，但是分散式的 Queue不能直接拿到，需要通過manager.get_task_queue()獲得Queue
    # 在windows 上跑不了 服務端
    import random, time, Queue
    from multiprocessing.managers import BaseManager
    
    # 傳送任務的佇列
    task_queue = Queue.Queue()
    # 接收結果的佇列
    result_queue = Queue.Queue()
    
    # 從BaseManager 繼承的QueueManager
    class QueueManager(BaseManager):
        pass
    
    # 把兩個queue註冊到網路
    QueueManager.register('get_task_queue', callable = lambda : task_queue)
    QueueManager.register('get_result_queue', callable = lambda : result_queue)
    # 繫結埠5000，驗證碼'abc'
    manager = QueueManager(address = ('', 5000), authkey = 'abc')
    # 啟動queue
    manager.start()
    # 獲得通過網路訪問的Queue的物件
    task = manager.get_task_queue()
    result = manager.get_result_queue()
    # 放入任務
    for i in range(10):
        n = i #random.randint(0, 10000)
        print 'Put task %d...' % n
        task.put(n)
    # 從result讀取結果
    print 'Try get results...'
    for i in range(10):
        r = result.get(timeout = 10)
        print 'Result: %s' % r
    # 關閉
    manager.shutdown()

else:
    # 客戶端
    import time, sys, Queue
    from multiprocessing.managers import BaseManager
    
    # 建立雷類似的QueueManager
    class QueueManager(BaseManager):
        pass
    # 由於這個QueueManager只能從網路拿到，所以註冊是提供名字
    QueueManager.register('get_task_queue')
    QueueManager.register('get_result_queue')
    # 連線到伺服器
    server_addr = '192.168.1.108'
    print 'Connect to server %s...' % server_addr
    # 埠和驗證碼要正確
    m = QueueManager(address = (server_addr, 5000), authkey = 'abc')
    # 從網路連線
    m.connect()
    # 獲取Queue物件
    task = m.get_task_queue()
    result = m.get_result_queue()
    # 從task去除任務，處理並返回結果
    for i in range(10):
        try:
            n = task.get(timeout = 5)
            print 'run task %d * %d...' % (n, n)
            r = '%d * %d = %d' % (n, n, n * n)
            time.sleep(1)
            result.put(r)
        except Queue.Empty:
            print 'task queue is empty'
    #處理結束
    print 'exit'