wiki:Pythonの短いソースコード集

Version 81 (modified by nakiki, 13 years ago) (diff)

--

Python:コードフラグメント

  • テキストをコピーして、適当なファイル名を作ります。
  • python test.pyのように実行してみてください。
  • 研究課題も記述しておきました。お試しあれ。

ちょっと長すぎるので、複数ページに分離しましょうよ。--sgk

更新pingを送信する

PythonSample/UpdatePing

タプル

a = (1, 2, 3)
a1, a2, a3 = a
print a1
print a2
print a3

リスト

list = [1, 2, 3, 4, 5, 6, 7]
for item in list:
  if item % 3 == 0:
    list.remove(item)
print list
  • 課題:リストのサイズが大きくなった場合削除に時間がかかるか?

リストに含まれるか

list = [1, 2, 3, 4, 5, 6, 7]
x = 5
if x in list:
  print '%d exists' % x

list = [[1, 2], [3, 4], [5, 6]]
x = [3, 4]
if x in list:
  print '[%d, %d] exists' % (x[0], x[1])

タプルに含まれるか

x = 3
list2 = (1, 2, 3, 4, 5, 6, 7)
if x in list:
  print '%d exists' % x

x = (3, 4)
list3 = ((1, 2), (3, 4), (5, 6))
if x in list3:
  print '(%d, %d) exists' % x

繰り返し動作

def catstr(x, y):
  return x + y
seq1 = ['a', 'b', 'c']
seq2 = ['1', '2', '3']
print map(catstr, seq1, seq2)

一時的な関数定義

seq1 = ['a', 'b', 'c']
seq2 = ['1', '2', '3']
print map(lambda x,y: x+y, seq1, seq2)
  • defで関数を定義しなくても簡単なものなら埋め込める。

区切りで文字分割

import string

s = '192.168.1.1'
a = string.split(s, '.')
print a

時刻の表示

import time

print time.strftime("%Y%m%d%H%M%S ", time.localtime())

空白除去

import string

str = 'dkdkdkd dkdkdkdk aaaaa   '
str1 = string.rstrip(str, ' ')
print str1

標準入力からの一行の読み取り

import string
import sys

msg = sys.stdin.readline()
print string.strip(msg)

商と余り

(a, r) = divmod(13, 5)
print a
print r

ランダム

import random

v = []
v +=  [random.randint(0, 99)]
v +=  [random.randint(0, 99)]
v +=  [random.randint(0, 99)]
v +=  [random.randint(0, 99)]
v +=  [random.randint(0, 99)]
print v
  • randintは二つめの引数の値を含む範囲の乱数を作ります。Pythonっぽくない。最近はrandrangeってのがあって、こっちを使う方がわかりやすいですよ。--sgk

メッセージダイジェスト

import sha

md = sha.new()
str = 'test messsage' * 100
md.update(str)
n = md.digest()
print (n,)
  • ビット数が長いshaの160bitを使ったほうが安全です。

辞書の表示

a = {'k1':100, 'k2':101, 'k3':102}
for k in a:
  print k, a[k]

基底クラスの呼び出し

class ServerProxy(xmlrpclib.ServerProxy):
  def __init__(self, addr):
    t = MyTCPTransport()
    xmlrpclib.ServerProxy.__init__(self, addr, t)
  • Pythonでは基底クラスのコンストラクタは自動的には呼び出されない。呼び出すのを忘れないように。

キュー

import Queue

q = Queue.Queue()

for n in range(5):
  q.put('message '+ str(n))

while not q.empty():
  print q.get()
  • スレッド動作も考慮しているので便利かな。

スレッドを作る

import threading
import time

def process():
  while True:
    print 'x1'
    time.sleep(3)

th = threading.Thread(None, process, None)
th.setDaemon(True)
th.start()

while True:
  print 'x0'
  time.sleep(3)

外部コマンドの実行

import os

result = os.system('ls -al')
print result

コマンド実行の出力の表示

import popen2

(child_stdout, child_stdin, child_stderr) = popen2.popen3('ls -al')
print child_stdout.read()

ソケット.TCPクライアント

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(10)
try:
  r = sock.connect(('127.0.0.1', 12345))
  sock.send('test message')
except socket.error, e:
  print 'Error: %s' % e

ソケット.TCPサーバ

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(('127.0.0.1', 12345))
sock.listen(1)
try:
  while True:
    (conn, sa) = sock.accept()
    msg = conn.recv(8192)
    print msg
    conn.close()
except socket.error, e:
  print 'Error: %s' % e

ソケット.TCPサーバ

import SocketServer

class Handler(SocketServer.StreamRequestHandler):
  def handle(self):
    while True:
      line = self.rfile.readline()
      if len(line) == 0:
        return
      self.wfile.write(line)

server = SocketServer.TCPServer(('127.0.0.1', 12345), Handler)
print 'listening:', server.socket.getsockname()
server.serve_forever()
  • pythonの標準のものを使うともっとスマート。スレッド動作させるクラスもあるよ。

UDPクライアント

try:
  sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  sock.sendto('test message', ('localhost', 12345))
except socket.error, e:
  print 'Error: %s' % e
~

RPCサーバ

import SimpleXMLRPCServer

class Test:
  def inc1(self, value):
    return value + 1

  def inc2(self, value):
    return value + 2

  def inc3(self, value):
    return value + 3

if __name__ == '__main__':
  server = SimpleXMLRPCServer.SimpleXMLRPCServer(('127.0.0.1', 12346))
  server.register_instance(Test())
  server.serve_forever()
  • 関数呼び出しのイメージで処理が進められるので便利。
  • 特別なプロトコル自分で作る必要がない。
  • 練習問題: このサーバを起動して、一度だけRPCクライアントとの通信後サーバを^Cで終了させてその直後にサーバをまた起動するとソケットがまだバインドされているというエラーがでてすぐに起動できない。さてどう改造したらうまく行くのでしょうか?

RPCクライアント

import xmlrpclib

server = xmlrpclib.ServerProxy('http://127.0.0.1:12346')
print server.inc1(1)
print server.inc2(1)
print server.inc3(1)
  • 関数を呼ぶだけです。

データの連結

s1 = 'sssssssss1'
s2 = 'sssssssss2'
s3 = 'sssssssss3'
s4 = 'sssssssss4'
l = [s1, s2, s3, s4]
print ''.join(l)
  • コピーを少なくしてデータを連結するため。

バッファ

s = '123456789abcdefgh'
x = buffer(s, 6, 2)
ss = str(x)
print ss
  • 長いデータの一部を取り出したり、見たりするときに便利。

文字列バッファ

import StringIO

f = StringIO.StringIO()
for n in range(4):
  f.write(str(n) + '\n')

f.seek(0)
print f.read()

バイナリデータを扱う

 packet = struct.pack("!HH", 500, 1)
 (l, x) = struct.unpack("!HH", packet[0:4])
 print l, x
  • ネットワークのパケット処理などに活躍します。

データの表示

print (data,)
  • printはタプルにして表示すると便利

ファイルのコピー

import shutil

shutil.copyfile('test.dat', 'test1.dat')

ディレクトリのコピー

import shutil

shutil.copytree('testdir', 'testdir2')

環境変数の表示

import os

for env in os.environ:
  print '%s=%s' % (env, os.environ[env])

一時ファイル

import tempfile

fname = tempfile.mktemp('.dat', dir='.')
fp = open(fname, 'w')
fp.write('test message')
fp.close()

CGI

import CGIHTTPServer
import BaseHTTPServer
import SimpleHTTPServer

def serveForever(addr, HandlerClass = CGIHTTPServer.CGIHTTPRequestHandler,
        ServerClass = BaseHTTPServer.HTTPServer):
  HandlerClass.protocol_version = 'HTTP/1.0'
  httpd = ServerClass(addr, HandlerClass)
  sa = httpd.socket.getsockname()
  print "Serving HTTP on %s:%d" % (sa[0], sa[1])
  httpd.serve_forever()

if __name__ == '__main__':
  serveForever(('127.0.0.1', 8000))

コマンド引数

import optparse

parser = optparse.OptionParser()
parser.add_option("-u",  "--user", dest="user", help= 'user name')
(options, args) = parser.parse_args()

if options.user:
  print options.user
else:
  print parser.print_help()
  • 自分で作らなくても便利なので最近利用しています。

gzipファイルの展開

import tarfile

tf = tarfile.TarFile.gzopen('test.dat.tar.gz', mode = 'r')
print tf.getmembers()
for l in tf.getmembers():
  tf.extract(l)

FTPクライアント

import ftplib

ftp = ftplib.FTP('yourhost')
ftp.login('user', 'xxxxx')
f = open('xxxx.exe', 'wb')
ftp.retrbinary('RETR xxxx.exe', f.write)
f.close()

HTTPクライアント

import httplib

h = httplib.HTTPConnection('www.yahoo.com')
h.request('GET', '/index.html')
r = h.getresponse()
if r.status == httplib.OK:
  data = r.read()
  print data
print 'Status=', r.status

HTTPサーバ

import BaseHTTPServer
import SimpleHTTPServer

addr = ('127.0.0.1', 8000)
Server = BaseHTTPServer.HTTPServer
handler = SimpleHTTPServer.SimpleHTTPRequestHandler

httpd = Server(addr, handler)
httpd.serve_forever()

メールクライアント

import smtplib

server = smtplib.SMTP('localhost')
fromaddr = 'xxx@xxx.xxx.xxx'
toaddr = 'yyy@yyy.yyy.yyy'
msg = 'test message'
server.sendmail(fromaddr, toaddr, msg)
server.quit()

簡単なデータベース

import dbm

db = dbm.open('test', 'rw')

if not db.has_key('key1'):
  db['key1'] = 'record1'
  print 'add record1'
else:
  print db['key1']
  • 複数プロセスでアクセスする場合はロックがいるのだろうか?

簡単なデータベース

import shelve

db = shelve.open('test.db')

if not db.has_key('key1'):
  db['key1'] = ('v1','v2','v3')
  print 'add new record'
else:
  print 'exists key1 record ', db['key1']
  • shelveを使うとpythonのタプル、リスト、辞書が格納できるのでとても便利。

シグナルハンドラ

import signal
import sys
import time

def handler(signum, frame):
  print 'signal %s' % signum
  sys.exit(signum)

signal.signal(signal.SIGTERM, handler)
while True:
  time.sleep(3)
  • メインスレッドでハンドラを設定するとファイルのクローズ処理などきちんとしてくれる。
  • 設定しないと、保存せずに終わったりする。ので便利。

CSVファイルの読み込み

import _csv

s = _csv.reader(file('test.txt', 'r'))
for ss in s:
  print ss
  • トークン解析をしなくて済む道具あることはすばらしい。
  • _csvにしないと動かないのはなぜ?