Doug Hellmann

do not

import os.path

for path in [ '/one/two/three', 
              '/one/two/three/',
              '/',
              '.',
              '']:
    print '"%s" : "%s"' % (path, os.path.split(path))

$ python ospath_split.py
"/one/two/three" : "('/one/two', 'three')"
"/one/two/three/" : "('/one/two/three', '')"
"/" : "('/', '')"
"." : "('', '.')"
"" : "('', '')"

import os.path

for path in [ '/one/two/three', 
              '/one/two/three/',
              '/',
              '.',
              '']:
    print '"%s" : "%s"' % (path, os.path.basename(path))

$ python ospath_basename.py
"/one/two/three" : "three"
"/one/two/three/" : ""
"/" : ""
"." : "."
"" : ""

import os.path

for path in [ '/one/two/three', 
              '/one/two/three/',
              '/',
              '.',
              '']:
    print '"%s" : "%s"' % (path, os.path.dirname(path))

$ python ospath_dirname.py
"/one/two/three" : "/one/two"
"/one/two/three/" : "/one/two/three"
"/" : "/"
"." : ""
"" : ""

import os.path

for path in [ 'filename.txt', 'filename', '/path/to/filename.txt', '/', '' ]:
    print '"%s" :' % path, os.path.splitext(path)

$ python ospath_splitext.py
"filename.txt" : ('filename', '.txt')
"filename" : ('filename', '')
"/path/to/filename.txt" : ('/path/to/filename', '.txt')
"/" : ('/', '')
"" : ('', '')

import os.path

paths = ['/one/two/three/four',
         '/one/two/threefold',
         '/one/two/three/',
         ]
print paths
print os.path.commonprefix(paths)

$ python ospath_commonprefix.py
['/one/two/three/four', '/one/two/threefold', '/one/two/three/']
/one/two/three

import os.path

for parts in [ ('one', 'two', 'three'),
               ('/', 'one', 'two', 'three'),
               ('/one', '/two', '/three'),
               ]:
    print parts, ':', os.path.join(*parts)

$ python ospath_join.py
('one', 'two', 'three') : one/two/three
('/', 'one', 'two', 'three') : /one/two/three
('/one', '/two', '/three') : /three

import os.path

for user in [ '', 'dhellmann', 'postgres' ]:
    lookup = '~' + user
    print lookup, ':', os.path.expanduser(lookup)

$ python ospath_expanduser.py
~ : /Users/dhellmann
~dhellmann : /Users/dhellmann
~postgres : /var/empty

import os.path
import os

os.environ['MYVAR'] = 'VALUE'

print os.path.expandvars('/path/to/$MYVAR')

$ python ospath_expandvars.py
/path/to/VALUE

import os.path

for path in [ 'one//two//three', 
              'one/./two/./three', 
              'one/../one/two/three',
              ]:
    print path, ':', os.path.normpath(path)

$ python ospath_normpath.py
one//two//three : one/two/three
one/./two/./three : one/two/three
one/../one/two/three : one/two/three

import os.path

for path in [ '.', '..', './one/two/three', '../one/two/three']:
    print '"%s" : "%s"' % (path, os.path.abspath(path))

$ python ospath_abspath.py
"." : "/Users/dhellmann/Documents/PyMOTW/in_progress/ospath"
".." : "/Users/dhellmann/Documents/PyMOTW/in_progress"
"./one/two/three" : "/Users/dhellmann/Documents/PyMOTW/in_progress/ospath/one/two/three"
"../one/two/three" : "/Users/dhellmann/Documents/PyMOTW/in_progress/one/two/three"

import os.path
import time

print 'File         :', __file__
print 'Access time  :', time.ctime(os.path.getatime(__file__))
print 'Modified time:', time.ctime(os.path.getmtime(__file__))
print 'Change time  :', time.ctime(os.path.getctime(__file__))
print 'Size         :', os.path.getsize(__file__)

$ python ospath_properties.py
File         : /Users/dhellmann/Documents/PyMOTW/in_progress/ospath/ospath_properties.py
Access time  : Sun Jan 27 15:40:20 2008
Modified time: Sun Jan 27 15:39:06 2008
Change time  : Sun Jan 27 15:39:06 2008
Size         : 478

import os.path

for file in [ __file__, os.path.dirname(__file__), '/', './broken_link']:
    print 'File        :', file
    print 'Absolute    :', os.path.isabs(file)
    print 'Is File?    :', os.path.isfile(file)
    print 'Is Dir?     :', os.path.isdir(file)
    print 'Is Link?    :', os.path.islink(file)
    print 'Mountpoint? :', os.path.ismount(file)
    print 'Exists?     :', os.path.exists(file)
    print 'Link Exists?:', os.path.lexists(file)
    print

$ ln -s /does/not/exist broken_link
$ python ospath_tests.py
File        : /Users/dhellmann/Documents/PyMOTW/in_progress/ospath/ospath_tests.py
Absolute    : True
Is File?    : True
Is Dir?     : False
Is Link?    : False
Mountpoint? : False
Exists?     : True
Link Exists?: True

File        : /Users/dhellmann/Documents/PyMOTW/in_progress/ospath
Absolute    : True
Is File?    : False
Is Dir?     : True
Is Link?    : False
Mountpoint? : False
Exists?     : True
Link Exists?: True

File        : /
Absolute    : True
Is File?    : False
Is Dir?     : True
Is Link?    : False
Mountpoint? : True
Exists?     : True
Link Exists?: True

File        : ./broken_link
Absolute    : False
Is File?    : False
Is Dir?     : False
Is Link?    : True
Mountpoint? : False
Exists?     : False
Link Exists?: True

import os.path
import pprint

def visit(arg, dirname, names):
    print dirname, arg
    for name in names:
        subname = os.path.join(dirname, name)
        if os.path.isdir(subname):
            print '  %s/' % name
        else:
            print '  %s' % name
    # Do not recurse into .svn directory
    if '.svn' in names:
        names.remove('.svn')
    print

os.path.walk('..', visit, '(User data)')

$ python ospath_walk.py
.. (User data)
  .svn/
  ospath/

../ospath (User data)
  .svn/
  __init__.py
  ospath_abspath.py
  ospath_basename.py
  ospath_commonprefix.py
  ospath_dirname.py
  ospath_expanduser.py
  ospath_expandvars.py
  ospath_join.py
  ospath_normpath.py
  ospath_properties.py
  ospath_split.py
  ospath_splitext.py
  ospath_tests.py
  ospath_walk.py

import hashlib

from hashlib_data import lorem

h = hashlib.md5()
h.update(lorem)
print h.hexdigest()

$ python hashlib_md5.py
c3abe541f361b1bfbbcfecbf53aad1fb

import hashlib

from hashlib_data import lorem

h = hashlib.sha1()
h.update(lorem)
print h.hexdigest()

$ python hashlib_sha1.py
ac2a96a4237886637d5352d606d7a7b6d7ad2f29

import hashlib
import sys

try:
    hash_name = sys.argv[1]
except IndexError:
    print 'Specify the hash name as the first argument.'
else:
    try:
        data = sys.argv[2]
    except IndexError:    
        from hashlib_data import lorem as data

    h = hashlib.new(hash_name)
    h.update(data)
    print h.hexdigest()

$ python hashlib_new.py sha1
ac2a96a4237886637d5352d606d7a7b6d7ad2f29
$ python hashlib_new.py sha256
88b7404fc192fcdb9bb1dba1ad118aa1ccd580e9faa110d12b4d63988cf20332
$ python hashlib_new.py sha512
f58c6935ef9d5a94d296207ee4a7d9bba411539d8677482b7e9d60e4b7137f68d25f9747cab62fe752ec5ed1e5b2fa4cdbc8c9203267f995a5d17e4408dccdb4
$ python hashlib_new.py md5   
c3abe541f361b1bfbbcfecbf53aad1fb
import hashlib

from hashlib_data import lorem

h = hashlib.md5()
h.update(lorem)
all_at_once = h.hexdigest()

def chunkize(size, text):
    "Return parts of the text in size-based increments."
    start = 0
    while start < len(text):
        chunk = text[start:start+size]
        yield chunk
        start += size
    return

h = hashlib.md5()
for chunk in chunkize(64, lorem):
    h.update(chunk)
line_by_line = h.hexdigest()

print 'All at once :', all_at_once
print 'Line by line:', line_by_line
print 'Same        :', (all_at_once == line_by_line)

$ python hashlib_update.py
All at once : c3abe541f361b1bfbbcfecbf53aad1fb
Line by line: c3abe541f361b1bfbbcfecbf53aad1fb
Same        : True

 $ make
gcc  -DNDEBUG -g -O3  -I/Users/dhellmann/Devel/AthensDocket/bin/../include/python2.5 
-I/Users/dhellmann/Devel/AthensDocket/bin/../lib/python2.5/config -DPACKAGE_NAME=\"psycopg\" 
-DPACKAGE_TARNAME=\"psycopg\" -DPACKAGE_VERSION=\"1.1.21\" -DPACKAGE_STRING=\"psycopg\ 1.1.21\" 
-DPACKAGE_BUGREPORT=\"psycopg@lists.initd.org\" -DHAVE_LIBCRYPTO=1 -DHAVE_ASPRINTF=1  
-I/Library/PostgreSQL8/include -I/Library/PostgreSQL8/include/postgresql/server 
-I../egenix-mx-base-3.0.0/mx/DateTime/mxDateTime -DHAVE_PQFREENOTIFY -DNDEBUG -D_REENTRANT 
-D_GNU_SOURCE -DPOSTGRESQL_MAJOR=8 -DPOSTGRESQL_MINOR=2   -c ././module.c -o ./module.o
In file included from ././module.c:33:
././module.h:30:20: error: Python.h: No such file or directory
$ CPPFLAGS=-I/Library/Frameworks/Python.framework/Headers \
> LDFLAGS=-L/Developer/SDKs/MacOSX10.5.sdk/usr/lib \
> ./configure --with-postgres-libraries=/Library/PostgreSQL8/lib \
> --with-postgres-includes=/Library/PostgreSQL8/include \
> --with-mxdatetime-includes=../egenix-mx-base-3.0.0/mx/DateTime/mxDateTime
import threading

def worker():
    """thread worker function"""
    print 'Worker'
    return

threads = []
for i in range(5):
    t = threading.Thread(target=worker)
    threads.append(t)
    t.start()

$ python threading_simple.py
Worker
Worker
Worker
Worker
Worker

import threading

def worker(num):
    """thread worker function"""
    print 'Worker:', num
    return

threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

$ python threading_simpleargs.py
Worker: 0
Worker: 1
Worker: 2
Worker: 3
Worker: 4

import threading
import time

def worker():
    print threading.currentThread().getName(), 'Starting'
    time.sleep(2)
    print threading.currentThread().getName(), 'Exiting'

def my_service():
    print threading.currentThread().getName(), 'Starting'
    time.sleep(3)
    print threading.currentThread().getName(), 'Exiting'

t = threading.Thread(name='my_service', target=my_service)
w = threading.Thread(name='worker', target=worker)
w2 = threading.Thread(target=worker) # use default name

w.start()
w2.start()
t.start()

$ python threading_names.py
worker Starting
Thread-1 Starting
my_service Starting
worker Exiting
Thread-1 Exiting
my_service Exiting

import logging
import threading
import time

logging.basicConfig(level=logging.DEBUG,
                    format='[%(levelname)s] (%(threadName)-10s) %(message)s',
                    )

def worker():
    logging.debug('Starting')
    time.sleep(2)
    logging.debug('Exiting')

def my_service():
    logging.debug('Starting')
    time.sleep(3)
    logging.debug('Exiting')

t = threading.Thread(name='my_service', target=my_service)
w = threading.Thread(name='worker', target=worker)
w2 = threading.Thread(target=worker) # use default name

w.start()
w2.start()
t.start()

$ python threading_names_log.py
[DEBUG] (worker    ) Starting
[DEBUG] (Thread-1  ) Starting
[DEBUG] (my_service) Starting
[DEBUG] (worker    ) Exiting
[DEBUG] (Thread-1  ) Exiting
[DEBUG] (my_service) Exiting

import threading
import time

def daemon():
    print 'Starting:', threading.currentThread().getName()
    time.sleep(2)
    print 'Exiting :', threading.currentThread().getName()

d = threading.Thread(name='daemon', target=daemon)
d.setDaemon(True)

def non_daemon():
    print 'Starting:', threading.currentThread().getName()
    print 'Exiting :', threading.currentThread().getName()

t = threading.Thread(name='non-daemon', target=non_daemon)

d.start()
t.start()

$ python threading_daemon.py
Starting: daemon
Starting: non-daemon
Exiting : non-daemon

import threading
import time

def daemon():
    print 'Starting:', threading.currentThread().getName()
    time.sleep(2)
    print 'Exiting :', threading.currentThread().getName()

d = threading.Thread(name='daemon', target=daemon)
d.setDaemon(True)

def non_daemon():
    print 'Starting:', threading.currentThread().getName()
    print 'Exiting :', threading.currentThread().getName()

t = threading.Thread(name='non-daemon', target=non_daemon)

d.start()
t.start()

d.join()
t.join()

$ python threading_daemon_join.py
Starting: daemon
Starting: non-daemon
Exiting : non-daemon
Exiting : daemon

import threading
import time

def daemon():
    print 'Starting:', threading.currentThread().getName()
    time.sleep(2)
    print 'Exiting :', threading.currentThread().getName()

d = threading.Thread(name='daemon', target=daemon)
d.setDaemon(True)

def non_daemon():
    print 'Starting:', threading.currentThread().getName()
    print 'Exiting :', threading.currentThread().getName()

t = threading.Thread(name='non-daemon', target=non_daemon)

d.start()
t.start()

d.join(1)
print 'd.isAlive()', d.isAlive()
t.join()

$ python threading_daemon_join_timeout.py
Starting: daemon
Starting: non-daemon
Exiting : non-daemon
d.isAlive() True

import random
import threading
import time

def worker():
    """thread worker function"""
    t = threading.currentThread()
    pause = random.randint(1,5)
    print 'Starting:', t.getName(), 'sleeping', pause
    time.sleep(pause)
    print 'Ending  :', t.getName()
    return

for i in range(3):
    t = threading.Thread(target=worker)
    t.setDaemon(True)
    t.start()

main_thread = threading.currentThread()
for t in threading.enumerate():
    if t is main_thread:
        continue
    print 'Joining :', t.getName()
    t.join()

$ python threading_enumerate.py
Starting: Thread-1 sleeping 2
Starting: Thread-2 sleeping 5
Starting: Thread-3 sleeping 2
Joining : Thread-1
Ending  : Thread-1
Joining : Thread-3
Ending  : Thread-3
Joining : Thread-2
Ending  : Thread-2

import threading

class MyThread(threading.Thread):

    def run(self):
        print 'MyThread:', self.getName()
        return

for i in range(5):
    t = MyThread()
    t.start()

$ python threading_subclass.py
MyThread: Thread-1
MyThread: Thread-2
MyThread: Thread-3
MyThread: Thread-4
MyThread: Thread-5

import threading
import time

def delayed():
    print 'Worker running', threading.currentThread().getName()
    return

t1 = threading.Timer(3, delayed)
t1.setName('t1')
t2 = threading.Timer(3, delayed)
t2.setName('t2')

print 'Starting timers'
t1.start()
t2.start()

print 'Waiting before canceling', t2.getName()
time.sleep(2)
print 'Canceling', t2.getName()
t2.cancel()
print 'Main thread done'

$ python threading_timer.py
Starting timers
Waiting before canceling t2
Canceling t2
Main thread done
Worker running t1

import logging
import threading
import time

logging.basicConfig(level=logging.DEBUG,
                    format='%(asctime)s (%(threadName)-10s) %(message)s',
                    )

def wait_for_event(e):
    """Wait for the event to be set before doing anything"""
    logging.debug('wait_for_event starting')
    e.wait()
    logging.debug('e.isSet()->%s', e.isSet())

def wait_for_event_timeout(e, t):
    """Wait t seconds and then timeout"""
    logging.debug('wait_for_event_timeout starting')
    e.wait(t)
    logging.debug('e.isSet()->%s', e.isSet())

e = threading.Event()
t1 = threading.Thread(name='block', 
                      target=wait_for_event,
                      args=(e,))
t1.start()

t2 = threading.Thread(name='non-block', 
                      target=wait_for_event_timeout, 
                      args=(e, 2))
t2.start()

logging.debug('Waiting before calling Event.set()')
time.sleep(3)
e.set()
logging.debug('Event is set')

$ python threading_event.py
2008-01-13 13:25:02,514 (block     ) wait_for_event starting
2008-01-13 13:25:02,525 (non-block ) wait_for_event_timeout starting
2008-01-13 13:25:02,536 (MainThread) Waiting before calling Event.set()
2008-01-13 13:25:04,526 (non-block ) e.isSet()->False
2008-01-13 13:25:05,563 (MainThread) Event is set
2008-01-13 13:25:05,564 (block     ) e.isSet()->True

import logging
import random
import threading
import time

logging.basicConfig(level=logging.DEBUG,
                    format='(%(threadName)-10s) %(message)s',
                    )

class Counter(object):
    def __init__(self, start=0):
        self.lock = threading.Lock()
        self.value = start
    def increment(self):
        self.lock.acquire()
        try:
            logging.debug('Acquired lock')
            self.value = self.value + 1
        finally:
            self.lock.release()

def worker(c):
    for i in range(3):
        pause = random.random()
        logging.debug('Sleeping %0.02f', pause)
        time.sleep(pause)
        c.increment()
    logging.debug('Done')

counter = Counter()
for i in range(5):
    t = threading.Thread(target=worker, args=(counter,))
    t.start()

logging.debug('Waiting for worker threa

Doug Hellmann

Tuesday, January 29, 2008

Back issues of Python Magazine

Monday, January 28, 2008

ProctorTicket

Sunday, January 27, 2008

PyMOTW: os.path

PyCon 2008

Saturday, January 26, 2008

January issue of Python Magazine now available for download

Sunday, January 20, 2008

PyMOTW: hashlib

Django with PostgreSQL on Mac OS X Leopard

Saturday, January 19, 2008

Python development tools you can't live without

Sunday, January 13, 2008

PyMOTW: threading