Please note that as of October 24, 2014, the Nokia Developer Wiki will no longer be accepting user contributions, including new entries, edits and comments, as we begin transitioning to our new home, in the Windows Phone Development Wiki. We plan to move over the majority of the existing entries. Thanks for all your past and future contributions.

Archived:Python on Symbian/14. Advanced Network Programming

From Wiki
Jump to: navigation, search

Archived.pngAquivado: Este artigo foi arquivado, pois o conteúdo não é mais considerado relevante para se criar soluções comerciais atuais. Se você achar que este artigo ainda é importante, inclua o template {{ForArchiveReview|escreva a sua justificativa}}.

All PySymbian articles have been archived. PySymbian is no longer maintained by Nokia and is not guaranteed to work on more recent Symbian devices. It is not possible to submit apps to Nokia Store.

Original Author: Marcelo Barros

Article Metadata
Code ExampleArticle
Created: marcelobarrosalmeida (01 Oct 2010)
Last edited: hamishwillee (08 May 2013)

This chapter extends the concepts discussed in Chapter 9, employing high level libraries to create advanced connected applications.



While Chapter 9 is useful to acquire a basic understanding of TCP/IP, and discusses how client-server applications can be constructed using the socket library, it does not explore the Python libraries available for developing networked programs.

In this chapter, some important libraries like urllib, xmlrpclib, JSON and Beautiful Soup are discussed and illustrated using fully functional examples written in Python for Symbian. We also present some techniques for inter-process communication and explain how to create multi-threaded programs.

HTTP Principles and urllib module


urllib is a versatile Python module for fetching data across the Internet. It has several interesting features:

  • Opening URLs with an interface similar to that one found in file operations
  • Functions for processing URLs, like escaping HTML and parameter processing
  • Proxy and HTTP basic authentication support

Powerful programs can be created with urllib. For instance, suppose you want to fetch the content of a page fromWiki Homeand save it into local file wikiforumnokia.html. This can be performed by urllib using a few lines of code:

import urllib
# returns a file like interface
furl = urllib.urlopen("")
# reading the "file"
contents =
# saving the page contents
flocal = open("wikiforumnokia.html","wt")

Or, if you prefer, the urlretrieve() method can achieve this with just two lines of code:

import urllib

urlopen() performs an HTTP GET request and fetches the contents of the desired page, stripping the HTTP header. If additional parameters are necessary in your request, they may be added to the URL like a typical URL GET request.

import urllib
params = urllib.urlencode({'name': 'My name is Bond', 'phone':'007007'})
url = "" + params
print url

The output is an URL with all parameters encoded, as you can see in the address bar when searching at Yahoo! or Google.


It is possible to simulate an HTTP POST request as well. POST requests do not use the URL to encode the parameters. Instead, parameters are included in the body of the request. Forms are a good example of POST requests where all form parameters are hidden inside the html body.

We can access with the same arguments as previously, but now using POST, as demonstrated in the following code snippet:

import urllib
params = urllib.urlencode({'name': 'My name is Bond', 'phone':'007007'})
result = urllib.urlopen("", params).read()

In this case, an additional parameter must be supplied to urlopen(), indicating the POST request parameters.

Accessing Wordpress statistics

As an example, urllib may be used to retrieve information about blog views and post views from wordpress statistics ( It is necessary to use an api_key ( and to create an appropriate HTTP GET request (see for more details about HTTP requests).

Blog views may be fetched with the following URL:

All post views may be fetched with the following URL:

Wordpress can send the response in CSV (comma separated values) or XML. I will use CSV, parsing just a few fields. It is necessary to use a smarter strategy to avoid problems with commas in post titles, for example (or switch to XML). Moreover, you may have a lot of headaches when transforming HTML in Unicode (avoided here as well).

The code is as follows and it is not difficult to understand after this urllib lesson.

# Wordpress Stats demo 
import urllib
from appuifw import *
import e32
class WPStats(object):
""" This classe uses urllib for accessing wordpress blog statistics.
Only blogs hosted at may be used.

def __init__(self,api_key,blog_uri,blog_id=0,max_days=30):
""" Init WPStats parmeters.
Please use:
api_key: copy it from
blog_uri: your blog uri (
max_days: all accesses will provided statistics for the last max_days

self.api_key = api_key
self.blog_uri = blog_uri
self.blog_id = blog_id
self.max_days = max_days
def __request_stats(self,custom_params):
""" Common request function. Additional parameters may be
encoded for GET using custom_params dictionary

params = {"api_key":self.api_key,
params.update(custom_params) # add custom_params values to params
f = urllib.urlopen(self.STAT_URL + urllib.urlencode(params))
except Exception, e:
raise e
data = []
rsp =
if rsp:
# this split may fail for post title with "\n" on it - improve it
data = rsp.split("\n")[1:-1] # discard column names and last empty element
return data
def get_post_views(self,post_id = 0):
""" Get the number of views for a given post id or
number of views for all posts (post id = 0)
Response is an array of tuples like below:

params = {"table":"postviews"}
if post_id:
params['post_id'] = post_id
data = self.__request_stats(params)
res = []
for d in data:
# this split may fail for post title with "," on it
row = d.split(",")
return res
def get_blog_views(self):
""" Get the number of views
Response format is an array of tuples like below:

params = {"table":"view"}
data = self.__request_stats(params)
res = []
for d in data:
return res
class WPStatClient(object):
""" Get statistics from wordpress

def __init__(self):
self.lock = e32.Ao_lock()
app.title = u"WP Stats demo" = [(u"Get blog views", self.blog_views),
(u"Get post views", self.post_views),
(u"Exit", self.close_app)]
self.body = Listbox([(u"Please, update statistics",u"")])
app.body = self.body
app.screen = "normal"
self.wpstats = WPStats("put_api_key_here","")
def blog_views(self):
bv = self.wpstats.get_blog_views()
note(u"Impossible to get stats","error")
if bv:
items = []
for stat in bv:
u"Views:" + unicode(stat[1])))
def post_views(self):
pv = self.wpstats.get_post_views()
note(u"Impossible to get stats","error")
if pv:
items = []
for stat in pv:
u"PostID:"+unicode(stat[1]) + u" Views:"+unicode(stat[2])))
def close_app(self):
if __name__ == "__main__":

Figure 14.1 shows some screenshots:

Accessing public Qik streams from Symbian devices

Qik is a new and innovative service that allows you stream video live from your cell phone to the web. The videos can be shared with your friends and are available for viewing and downloading. In this section, we'll describe how to use urllib and JSON-RPC to create an API for browsing public streams, given a Qik account.

Object serialization using JSON

JSON is a lightweight data-interchange format, easily readable for humans and equally easily parsed and generated by machines. Several languages, including Python, have one or more JSON implementations. In this section, the Python library known as simplejson will be used. This library was also ported for Symbian devices and it is available for download from

Using simplejson, almost all basic types may be encoded as strings and sent over an Internet connection. Two simplejson methods (dumps() and loads()') are responsible for encoding data as a JSON representation and decoding it afterwards.

For instance, consider the following dictionary and its posterior serialization/deserialization:

import simplejson
d = { "name":"John Symbian", "age": 36, "weight":1.77, "devices":["N95", "E72", "N900"] }
ser = simplejson.dumps(d)
print len(ser)
print type(ser)
print ser
rd = simplejson.loads(ser)
print len(rd)
print type(rd)
print rd

The serialization output is a string, ready to be transmitted over TCP/IP connections:

>>> 81
>>> <type 'str'>
>>> '{"age": 36, "devices": ["N95", "E72", "N900"], "name": "John Symbian", "weight": 1.77}'

This string can be converted to a Python object again using the loads() method, as showed in the following output:

>>> 4
>>> <type 'dict'>
>>> {u'age': 36, u'weight': 1.77, u'name': u'John Symbian', u'devices': [u'N95', u'E72', u'N900']}

Qik API foundations

The current Qik API is based on JSON-RPC and REST (only available for the Qik-ly API and not used in this section). JSON-RPC Qik API uses HTTP for transfer data between client and Qik engine. For instance, the public streams of Qik user marcelobarrosalmeida can be retrieved with the following HTTP data:

POST /api/jsonrpc?apikey=YOUR_API_KEY HTTP/1.0
Content-Length: 80
Content-Type: application/json; charset=UTF-8
{"method": "","params": ["marcelobarrosalmeida"]}

The response is as follows:

HTTP/1.1 200 OK
Server: nginx/0.7.59
Date: Fri, 07 Aug 2009 14:31:06 GMT
Content-Type: text/json
Connection: close
Content-Length: 1706
X-Qik-Origin: 229
[[{"url": "", "live": false, "user_id": 365150, "small_thumbnail_url": "", "title": "Untitled",
"duration": 54, "created_at": "2009-07-31 13:23:30", "views": 7, "id": 2363785}, {"url": "",
"live": false, "user_id": 365150, "small_thumbnail_url": "",
"title": "Pint da Guinness", "duration": 30, "created_at": "2009-07-31 18:15:29", "views": 18, "id": 2366002},
{"url": "", "live": false, "user_id": 365150, "small_thumbnail_url": "", "title": "Trafalgar square",
"duration": 41, "created_at": "2009-07-31 14:01:39", "views": 11, "id": 2363998}, {"url": "",
"live": false, "user_id": 365150, "small_thumbnail_url": "",
"title": "Untitled", "duration": 55, "created_at": "2009-07-30 21:06:58", "views": 5, "id": 2357498}, {"url": "",
"live": false, "user_id": 365150, "small_thumbnail_url": "",
"title": "Untitled", "duration": 97, "created_at": "2009-07-30 19:54:10", "views": 7, "id": 2356796}, {"url": "",
"live": false, "user_id": 365150, "small_thumbnail_url": "",
"title": "Untitled", "duration": 61, "created_at": "2009-07-31 12:54:22", "views": 5, "id": 2363619}]]

The responses use JSON as well, and will vary depending on the required command (check for details).

Implementing Qik API in Python for Symbian

Using simplejson and urllib, it is possible to create an API for accessing Qik services from a mobile device. Since we need to include special HTTP headers, instead of using urllib.urlopen() we create an URL opener (using urllib.URLopener()), and add required headers to it:

urlopener = urllib.URLopener()
urlopener.addheaders = [('Host',''), ('Content-Type','application/json; charset=UTF-8')]

The desired remote procedure call can be reached using the open() method and a serialized version of your data. The simplejson method dumps() is used to accomplish this task.

import simplejson as json
data = json.dumps(data)
url = ''
f =,data)

Finally, using the file descriptor returned by open() and loads() to deserialize the data, it is possible to get the response and decode it:

data = json.loads([0]

The code can be run in Python for Symbian devices or PC, but you will need an API key to use it, which you can request from

# -*- coding: utf-8 -*-
# License: GPL3
import simplejson as json
import urllib
class QikApi(object):
""" Simple class for Qik videos proxy support

def __init__(self, api_key, qik_usr):
""" Create a new Qik API instance with given API key and user name

self.qik_url = '' + api_key
self.qik_usr = qik_usr
self.qik_id = -1
def __urlopener(self):
""" Return an urlopener with Qik required headers already set

urlopener = urllib.URLopener()
urlopener.addheaders = [('Host',''),
('Content-Type','application/json; charset=UTF-8')]
return urlopener
def __open(self,url,params=""):
""" Open a given URL using GET or POST and Qik headers

if params:
f = self.__urlopener().open(url,params) #post
f = self.__urlopener().open(url) #get
return f
def __qik_request(self,data):
""" Qik request. Encode data in json format, do the request and
decode json response

data = json.dumps(data)
f = self.__open(self.qik_url,data)
res = json.loads([0]
return res
def __check_id(self,qik_id):
""" Check if user ID was retrieved or not. If not, download it

if qik_id == -1:
if self.qik_id == -1:
self.qik_id = self.get_user_public_profile()[u'id']
qik_id = self.qik_id
return qik_id
def get_public_user_streams(self,usr=''):
""" Return all public stream for a given user
(or for the current user, if it not provided)

if not usr:
usr = self.qik_usr
data = {'method': '','params': [usr]}
return self.__qik_request(data)
def get_user_public_profile(self,usr=''):
""" Return public profile for a given user
(or for the current user, if it not provided)

if not usr:
usr = self.qik_usr
data = {'method': 'qik.user.public_profile','params': [usr]}
return self.__qik_request(data)
def get_user_public_detailed_profile(self,usr=''):
""" Return detailed public profile for a given user
(or for the current user, if it not provided)

if not usr:
usr = self.qik_usr
data = {'method': 'qik.user.public_detailed_profile','params': [usr]}
return self.__qik_request(data)
def get_user_followers(self,qik_id=-1):
""" Return the list of followers for a given user
(or for the current user, if it not provided)

qik_id = self.__check_id(qik_id)
data = {'method': 'qik.user.followers','params': [qik_id]}
return self.__qik_request(data)
def get_user_following(self,qik_id=-1):
""" Return the list of following for a given user
(or for the current user, if it not provided)

qik_id = self.__check_id(qik_id)
data = {'method': 'qik.user.following','params': [qik_id]}
return self.__qik_request(data)
def get_public_stream_info(self,vid_id):
""" Get detailed information about some public video

data = {'method': '','params': [vid_id]}
return self.__qik_request(data)

Demo application and source code

It is simple to create a demo application to show user streams and their Qik following/followers lists, each one in a different tab. To play the videos on a Flash-lite capable device, a local HTML file with links to the embedded video is created (see QIK_TEMPLATE variable) and the native web browser is called to show it. The user interface was created using the handy article found at Archived:PySymbian basic user interface app framework

You can see this application in action as follows:

The media player is loading...

Check for newer versions and updates. To run it, just copy all files (,,, to your memory card (into e:\Python) and use the Python interpreter to execute

# Qik view demo
import sys
import window
from appuifw import *
from qikapi import QikApi
import time
<html><head><meta http-equiv="Content-Type" content="application/vnd.wap.xhtml+xml; charset=utf-8" /><title>__TITLE__</title></head><body>
<object classid="clsid:d27cdb6e-ae6d-11cf-96b8-444553540000" codebase=",0,0,0" width="220" height="185" id="player" align="middle">
<param name="movie" value=";thumbnail=;size=false&amp;aplay=true&amp;autorew=false&amp;layout=small&amp;title=__TITLE__"/>
<param name="menu" value="false" />
<param name="quality" value="high" />
<param name="bgcolor" value="#999999" />
<embed src=";thumbnail=;size=false&amp;aplay=true&amp;autorew=false&amp;layout=small&amp;title=__TITLE__" menu="false" quality="high" bgcolor="#999999" width="220" height="185" name="player" align="middle" allowScriptAccess="sameDomain" type="application/x-shockwave-flash" pluginspage=""/>
class QikView(window.Application):
def __init__(self):
self.qik_usr = u""
self.qik_api = None = {'profile':[], 'streams':[], 'followers':[], 'following':[]}
# menus
streams_menu = [(u"Show stream",self.show_video)]
common_menu = [(u"Update",self.update),
# bodies
self.streams = Listbox([(u"Please, setup and update",u"")],self.show_video)
self.following = Listbox([(u"Please, setup and update",u"")])
self.followers = Listbox([(u"Please, setup and update",u"")])
u"Qik View",
def update(self):
if not self.qik_usr or not self.qik_api:
note(u"Please, setup the Qik user",u"error")
self.set_title(u"Updating profile...")['profile'] = self.qik_api.get_user_public_profile()
self.set_title(u"Updating streams...")['streams'] = self.qik_api.get_public_user_streams()
self.set_title(u"Updating followers...")['followers'] = self.qik_api.get_user_followers()
self.set_title(u"Updating following...")['following'] = self.qik_api.get_user_following()
note(u"Network error. Please, try again","error")
self.set_title(u"Qik View")
def update_bodies(self):
streams = []
followers = []
following = []
for s in['streams']:
h1 = s['title'] + (u" (%ds)" % s['duration'])
h2 = s['created_at']
for f in['followers']:
for f in['following']:
if streams:
self.streams.set_list([(u"No streams available",u"")])
if followers:
self.followers.set_list([(u"No followers available",u"")])
if following:
self.following.set_list([(u"No following available",u"")])
def setup(self):
usr = query(u"Qik user:","text",self.qik_usr)
if usr is not None:
self.qik_usr = usr
self.qik_api = QikApi(API_KEY,self.qik_usr)
def show_video(self):
# retrieve information about video
idx = self.streams.current()
if 'stream_info' not in['streams'][idx]:
vid =['streams'][idx][u'id']
self.lock_ui(u"Downloading stream info...")
try:['streams'][idx]['stream_info'] = self.qik_api.get_public_stream_info(vid)
note(u"Network error. Please, try again","error")
ret = True
ret = False
self.set_title(u"Qik View")
if ret:
tit =['streams'][idx]['stream_info'][u'title'].encode('utf-8')
fn =['streams'][idx]['stream_info'][u'filename'].encode('utf-8')
html_code = QIK_TEMPLATE.replace('__FILENAME__',fn).replace('__TITLE__',tit)
html_file = "html_" + time.strftime("%Y%m%d_%H%M%S", time.localtime()) + ".html"
fp = open(html_file,"wt")
note(u"Could not create HTML file","error")
viewer = Content_handler(self.refresh)
note(u"Can not open browser","error")
if __name__ == "__main__":
app = QikView()

Beautiful Soup and an HTML/XML parser

Beautiful Soup ( is a Python HTML/XML parser with many useful methods to collect data from pages or for navigating, searching, and modifying a parse tree. It is very flexible and can be executed on Symbian devices, creating interesting mobile applications. Since Beautiful Soup does not come bundled with a standard Python distribution, you will need to download it and extract the file to your project root. At the time of writing the version was the most recent.

Beautiful Soup basics

Suppose you have a large (and confusing) HTML file. It is possible to organize and indent it using the prettify() method. Just create a BeautifulSoup object and feed it the HTML as follows:

from BeautifulSoup import BeautifulSoup
html = u"""<html><body><h1 style="text-align:center">Heading 1</h1>
<p>Page content goes here.
<h2>And here.</h2></p><a href=""
alt="Croozeus link">Croozeus</a><br/>
<a href="">Python</a><br/>
<h1>The end.</h1></body>

soup = BeautifulSoup(html)
print soup.prettify()

The output is as follows:

<h1 style="text-align:center">
Heading 1
Page content goes here.
And here.
<a href="" alt="Croozeus link">
<br />
<a href="">
<br />
The end.

The parser tree is available with nice operations like findAll(). For instance, how about printing all the links on the page ? Just use a dictionary with all the tags you want as argument:

links = soup.findAll({'a':True})
for link in links:
print "-->", link['href'].encode('utf-8')

The output is as follows:


Or you may want to add the alt attribute to all links, modifying the parser tree, which is simple:

for link in links:
link['alt'] = link['href']
print soup.prettify()

The output is as follows:

<a href="" alt="">
<br />
<a href="" alt="">

As you can see, each link is like a dictionary and it is really straightforward to modify it.

The contents between <p> and </p> tags can be retrieved with a call to find(), which just looks for the first instance of tag <p> and returns an object that holds the contents between <p> and </p>. Using the contents attribute, you can iterate over an array of child elements. In this case, the first is a string and the second is a new parseable element of the tree:

p = soup.find('p')
print p.contents
print p.contents[0]
print p.contents[1]
print p.contents[1].contents

The output is as follows:

[u'Page content goes here.\n', <h2>And here.</h2>]
Page content goes here.
<h2>And here.</h2>
[u'And here.']

There are a complete set of functions to navigate through the tree. For instance, it is possible to start our search at first tag 'p' after locating its child with the following code:

p = soup.p
h2 = p.findChild()
print h2

The output is as follows:

<h2>And here.</h2>

Or start the search at first tag 'h1' but now looking for next siblings:

while h1:
print h1
h1 = h1.findNextSibling('h1')

The output is as follows:

<h1 style="text-align:center">Heading 1</h1>
<h1>The end.</h1>

Link checker application

Now it's time to use this knowledge in a new application: a basic link checker for Symbian devices. The idea is to download the contents of some URL and check all links inside it using Beautiful Soup and urllib.

There is a known problem related to fetching pages from Wikipedia using Python. Wikipedia does not accept the default urllib user agent, so you need to change it. The class LCOpener was created to solve this issue, defining a new user agent (Mozilla/5.0).

# Link Checker demo
import sys
# Try to import 'btsocket' as 'socket'
sys.modules['socket'] = __import__('btsocket')
except ImportError:
import socket
from BeautifulSoup import BeautifulSoup
import os
import e32
import urllib
import hashlib
from appuifw import *
class LCOpener(urllib.FancyURLopener):
""" For mediawiki it is necessary to change the http agent.

version = 'Mozilla/5.0'
class LinkChecker(object):
def __init__(self):
self.lock = e32.Ao_lock()
self.dir = "e:\\linkchecker"
if not os.path.isdir(self.dir):
self.apo = None
self.url = u'http://www.'
self.running = False
app.title = u"Link Checker"
app.screen = "normal" = [(u"Check URL",self.check_url),
(u"Exit", self.close_app)]
self.body = Text()
app.body = self.body
def close_app(self):
def sel_access_point(self):
""" Select and set the default access point.
Return the access point object if the selection was done or None if not

aps = socket.access_points()
if not aps:
note(u"No access points available","error")
return None
ap_labels = map(lambda x: x['name'], aps)
item = popup_menu(ap_labels,u"Access points:")
if item is None:
return None
apo = socket.access_point(aps[item]['iapid'])
return apo
def check_url(self):
if self.running:
note(u"There is a checking already in progress",u"info")
self.running = True
url = query(u"URL to check", "text", self.url)
if url is not None:
self.url = url
self.apo = self.sel_access_point()
if self.apo:
self.running = False
def run_checker(self):
self.body.add(u"* Downloading page: %s ...\n" % self.url)
fn = os.path.join(self.dir,'temp.html')
except Exception, e:
self.body.add(u"Could not download " + self.url)
self.body.add(u"* Parsing links ...\n")
page = open(fn,'rb').read()
soup = BeautifulSoup(page)
self.body.add(u"* BeautifulSoup error when decoding html. Aborted.")
tags = soup.findAll({'img':True,'a':True})
links = {}
bad_links = []
for n,tag in enumerate(tags):
if 'href' in tag:
link = tag['href']
elif 'img' in tag:
link = tag['src']
# just check external links
if link.startswith(u'http'):
# not handling internal links
link = link.split(u'#')[0]
# using a hash to avoid repeated links
h = hashlib.md5()
links[h.digest()] = link
nl = len(links)
for n,k in enumerate(links):
link = links[k]
msg = u"[%d/%d] Checking %s " % (n+1,nl,link)
(valid,info) = self.check_link(link.encode('utf-8'))
if valid:
msg = u"==> Passed\n"
msg = u"==> Failed: %s\n" % info
msg = u"* Summary: %d links (%d failed)\n" % (nl,len(bad_links))
for link in bad_links:
self.body.add(u"==> %s failed\n" % link)
self.body.add(u"* Finished")
def check_link(self,link):
""" Check if link (encoded in utf-8) exists.
Return (True,'') or (False,'error message')

page = LCOpener().open(link)
except Exception, e:
return (False,unicode(repr(e)))
return (True,u'')
lc = LinkChecker()

Multi-threaded programming

It is impossible to talk about network programming without discuss multi-threading.

Many servers are implemented using multi-threading strategies, spawning new threads for each client connection that they receive. Multi-threaded programs require specific inter process communications like semaphores, critical sections and queues. Python support for multi-threading is provided by the threading module[1] and queues can be created using the Queue module, both available in Python for Symbian. They will be used in this chapter for constructing multi-threaded clients and servers.


Threads can be created from functions or Thread objects. For instance, in the following example a new thread is created to show n messages, waiting m seconds between two consecutive messages.

# Multi-thread demo1
from threading import Thread
from time import sleep
def mythread(n,m):
for i in xrange(n):
print "-->",i
t = Thread(target=mythread,args=(5,2))
Figure 14.3: Multi-thread demo result

Since you have created the function with your thread code, it is necessary to create a Thread object using target to indicate the thread and args to specify the thread parameters. Thus, it is possible to start a new thread from the main thread (your script) just by calling the method start(). The join() method is used to block the main thread while the dispatched runs, avoiding any premature termination of the main thread.

If you want to use classes, it is necessary to create a new class using Thread as the base class and defining your thread code inside the run() method, as follows:

# Multi-thread demo2
from threading import Thread
from time import sleep
class MyThread(Thread):
def __init__(self,n,m):
self.n = n
self.m = m
def run(self):
for i in xrange(self.n):
print "-->",i
t = MyThread(5,2)


Multi-threaded applications need to use synchronization mechanisms to control access to shared resources. The following example shows how to use a semaphore to control access to a list that can be modified by different threads:

# Semaphore demo
from threading import Thread, Semaphore
from time import sleep
from random import randint
class MyThread(Thread):
resource = []
def __init__(self,tid,s):
self.tid = tid
self.sema = s
def insert(self,v):
def run(self):
for i in xrange(5):
s = Semaphore(1)
tsks = [ MyThread(n,s) for n in xrange(5) ]
[ t.start() for t in tsks ]
[ t.join() for t in tsks ]
print MyThread.resource
Figure 14.4: Semaphore demo result

The semaphore object is created with a resource-counting object as argument, indicating the number of resources available. In this case, only one resource is used (a global list) and we avoid any concurrent access to it by using a semaphore. The insert() function can be understood as a critical section, and all code between acquire() and release() methods will be protected. Successive calls to acquire() without previous calls to release() will block the calling thread until the resource becomes available.


Python implements multi-producer, multi-consumer queues via the Queue module, which is available for Python for Symbia. The put() and get() methods are the basic Queue methods, inserting and removing objects into/from the Queue. Both methods support timeouts and can block or not. In the next code snippet, several worker threads are created. These threads receive their jobs using a Queue object:

# Queue demo
from threading import Thread
from Queue import Queue
from random import randint
from time import sleep
def worker_thread(ti,q):
while True:
job = q.get()
if job == "exit":
print "[%d] exit" % ti
print "[%d] New job: %s" % (ti,job)
q = Queue()
max_jobs = 12
max_threads = 3
# creating and starting all threads
threads = [ Thread(target=worker_thread,args=(x,q)) for x in xrange(max_threads) ]
[ t.start() for t in threads ]
# dispatching jobs
[ q.put("Job%d" % x) for x in xrange(max_jobs) ]
# sending exit
[ q.put("exit") for x in xrange(max_threads) ]
# waiting all threads
[ t.join() for t in threads ]

A possible output could be:

[0] New job: Job0
[1] New job: Job1
[2] New job: Job2
[2] New job: Job3
[2] New job: Job4
[0] New job: Job5
[0] New job: Job6
[0] New job: Job7
[2] New job: Job8
[1] New job: Job9
[0] New job: Job10
[2] New job: Job11
[2] exit
[1] exit
[0] exit

You may also want to check the task_done() and join() Queue methods (available since Python 2.5) for a simpler way to create worker threads.

Multi -threaded page download example

In this example, we will use threads to concurrently download several web pages. You just need to create a file with all pages that you want to download, as follows:

In this case, the last page was intentionally mistyped to test when the download fails. A queue is used to send progress messages from threads to the main user interface. All files will be saved into "e:\python" using the page name as a file name. The code is as follows:

# Multi-thread page downloader
import urllib
from threading import Thread
from Queue import Queue
from appuifw import *
import os
import e32
class get_page(Thread):
def __init__(self,url,qlog):
self.url = url
self.fname = os.path.join(u"e:\\python\\",url[url.rfind('/')+1:])
self.qlog = qlog
def add_msg(self,msg):
# send a message to UI
def run(self):
self.add_msg(u"Saving %s into %s" % (self.url,self.fname))
self.add_msg(u"Error downloading %s" % (self.url))
self.add_msg(u"%s finished" % (self.url))
class mt_page_download(object):
def __init__(self):
self.lock = e32.Ao_lock()
self.qlog = Queue()
app.title = u"MT Demo"
app.screen = "normal" = [(u"Load URL list", self.load_list),
(u"Exit", self.close_app)]
app.body = Text()
self.lst = u"e:\\python\\urls.txt"
self.running = True
def load_list(self):
lst = query(u"Path to URL list",'text',self.lst)
if lst is not None:
self.lst = lst
def create_threads(self):
urls = open(self.lst,'rt').readlines()
urls = [ url.replace('\n','') for url in urls if len(url) ]
urls = [ url.strip() for url in urls if len(url.strip()) ]
self.add_msg(u"Could not open %s" % self.lst)
self.add_msg(u"Creating threads for %s" % self.lst)
[ get_page(url,self.qlog).run() for url in urls ]
def close_app(self):
self.running = False
def add_msg(self,msg):
app.body.add(msg + u"\u2029")
def run(self):
while self.running:
# create a loop and check to incoming messages
msg = self.qlog.get(True,1)
mt = mt_page_download()

Recreating a Python shell with multi-threading support

Some scripts in this section will lock if you try to run the code in Python for Symbian shells from series 1.4 or before 1.9.1. Those Python for Symbian shell were created using ensymble with {{{1}}}, but for proper multi-threading and socket usage you need a shell with {{{1}}}, and you'll need to recreate the sis file, changing the package mode.

Note: This topic may be considered out of scope but since it is extremely important to run multi-threaded programs that use sockets, it will described briefly here. You should also check for more information.

You will need a copy of the current shell file, available in the Python for Symbian source code as the file ext\amaretto\scriptshell\ Extract and copy it into a directory, for instance c:\temp\pyshellmt\. Do not use spaces in any directory name.

Using the application Ensymble GUI you can recreate your shell. Ensymble GUI is available from the PythonForSymbian 2.0.0 installation package. Fill all fields as shown in the next figure.

  • We recommend that you add some additional platform security capabilities to your shell (if you're unsure about what this means, further information about Symbian platform security is available in Chapter 15). In this example we used:
  • Press Create and your new shell will be created inside c:\temp\pyshellmt\.
  • Install and use it for running all examples in this section.
Figure 14.6: Creating a new Python shell


The xmlrpclib module is also available for Python for Symbian, allowing communication between an application and several XML-RPC based services, like Wordpress, Flickr and Drupal. Using xmlrpclib it is possible to call remote methods (with parameters) and get back structured responses.

The following code snippet will show how to retrieve a list of recent posts from a Wordpress blog account. Before you run it, replace put_user_here and put_pass_here with blog credentials with administration rights, and check if your blog allows XML-RPC connections. Moreover, do not forget to use your own blog URL, in general following the format http://your_blog_domain/xmlrpc.php.

For further explanations about the methods and parameters available, please consult the documentation at

# XML-RPC demo
import xmlrpclib
blog = ''
server = xmlrpclib.ServerProxy(blog)
rposts = server.metaWeblog.getRecentPosts(0,"put_user_here","put_pass_here", 5)
for post in rposts:
print "-> %s" % post['title']

The output is as follows:

-> I think you already know but I will say ...
-> I update the latest Wordmobi version to ...
-> Nice things happens when you share :) Th...
-> Returning to Python 1.9.6 and new Wordmobi 0.9.3
-> Python 1.9.7 is crashing wordmobi due to an error in mktime() function.

The list of categories can be retrieved using the method wp.getCategories() (see for documentation):

# XML-RPC demo
cats = server.wp.getCategories(0,"put_user_here","put_pass_here")
for c in cats:
print c['categoryName']


New Versions

Recent services such as Flickr will require more advanced authentication methods. Please check the documentation for individual services before using them.

Express post demo application

In the next application, xmlrpclib is used to add small posts to a Wordpress-based blog. It is necessary to submit blog domain and access credentials, as cited in the previous section. If want to develop a new application for Wordpress, there is a xmlrpclib wrapper called Wordpresslib ( that can be useful.

# Wordpress demo
import socket
from appuifw import *
import xmlrpclib
import e32
WP_BLOG = "http://YOUR_BLOG_DOMAIN/xmlrpc.php"
class ExpressPost(object):
def __init__(self):
self.lock = e32.Ao_lock()
app.title = u"Express Post"
app.screen = "normal"
self.title = u""
self.contents = u"" = [(u"Send post",self.send_post),
(u"Exit", self.close_app)]
self.body = Listbox([(u"",u"")],self.edit)
app.body = self.body = xmlrpclib.ServerProxy(WP_BLOG)
def edit(self):
idx = self.body.current()
if idx == 0:
title = query(u"Title", "text", self.title)
if title is not None:
self.title = title
elif idx == 1:
contents = query(u"Contents", "text", self.contents)
if contents is not None:
self.contents = contents
def update(self):
def send_post(self):
"Adding a new post to the blog"
post = { 'title' : self.unicode_to_utf8(self.title),
'description' : self.unicode_to_utf8(self.contents)}
# use 0 instead 1 at the end if you want a draft post,WP_USER,WP_PASS,post,1)
note(u"Could not send the post!","error")
note(u"Post published!","info")
def unicode_to_utf8(self,s):
"Converting string to the default encoding of wordpress"
return s.encode('utf-8')
def close_app(self):

Source Code

All examples shown in this chapter are available for download from File:PythonOnSymbianBookExampleCode


If you are reading a printed version of this book, the following links may be useful:

  1. The threading module creates threads, not processes. If you want to create processes, please check the multiprocessing module


In this chapter, we discussed how to create fully-functional networked programs using important libraries such as urllib, xmlrpclib, JSON and Beautiful Soup. Multi-threaded applications and interprocess communications were briefly described too, since they are frequently found in networked applications.

Licence icon cc-by-sa 3.0-88x31.png© 2010 Symbian Foundation Limited. Portions copyright Bogdan Galiceanu, Hamish Willee, Marcelo Barros de Almeida, Mike Jipping, Pankaj Nathani and others in wiki document history list. This document is licensed under the Creative Commons Attribution-Share Alike 2.0 license. See for the full terms of the license.
Note that this content was originally hosted on the Symbian Foundation developer wiki.

This page was last modified on 8 May 2013, at 06:07.
520 page views in the last 30 days.