Differences between revisions 150 and 151

Python

Python is a programming language that lets you work more quickly and integrate your systems more effectively.

Links:

Python3

Install from SlackBuild

cd /tmp
wget http://slackbuilds.org/slackbuilds/13.1/development/python3.tar.gz
tar xvzf python3.tar.gz
cd python3
wget http://python.org/ftp/python/3.1.3/Python-3.1.3.tar.bz2
./python3.SlackBuild
installpkg /tmp/python3-3.1.1-i486-1_SBo.tgz

UTF-8

At the start of source code files:

   1 # -*- coding: utf-8 -*-
   2 print u"Olá mundo" # hello world in portuguese

Letter á is encoded as hexadecimal 0xC3A1 in UTF-8 and as 0x00E1 in UTF-16. http://www.fileformat.info/info/unicode/char/e1/index.htm

Letter ú is encoded as hexadecimal 0xC3BA in UTF-8, as 0x00FA in UTF-16 and as 0xFA in latin-1/ISO8859-1/CP-1252.

   1 #utf-8 data
   2 utf81 ='\xc3\xba' 
   3 # convert from utf-8 to unicode
   4 unic = utf81.decode('utf-8') 
   5 
   6 for c in unic:
   7     print '%04X'%(ord(c))
   8 # convert from unicode to latin1/ISO8859-1 CP-1252
   9 lat1=unic.encode('latin1') 
  10 for c in lat1:
  11     print '%02X'%(ord(c))

Utf-8 and Unicode

   1 utf81='\xe1\xb7\x97' #utf-8 data
   2 
   3 unic = utf81.decode('utf-8') #converts from utf-8 to unicode (utf-16)
   4 for c in unic:
   5     print '%04X'%(ord(c))
   6 type(unic) # unicode, 2 bytes per char
   7 type(utf81) # str, 1 bytes per char

Time and date

   1 import time
   2 # get seconds since epoch until now in UTC to a string year-month-dayThour:minute:second
   3 strutc = time.strftime('%Y-%m-%dT%H:%M:%S', time.gmtime() ) 
   4 # get seconds since epoch until now  in localtime to a string  year-month-dayThour:minute:second
   5 strlocal = time.strftime('%Y-%m-%dT%H:%M:%S', time.localtime() ) 
   6 # number seconds since epoch
   7 nrSeconds = time.mktime(time.gmtime())
   8 nrSeconds = time.time() #usually is UTC on all OS
   9 # from timestamp to string date
  10 import datetime
  11 print(datetime.datetime.fromtimestamp(1284101485).strftime('%Y-%m-%d %H:%M:%S'))
  12 #
  13 def toUTCDateStr(timestamp):
  14     return datetime.datetime.utcfromtimestamp( timestamp ).strftime('%Y-%m-%d %H:%M:%S ')
  15 # timedelta
  16 import datetime
  17 x=datetime.datetime.fromtimestamp(1284101485)
  18 nowx=datetime.datetime.now()                    
  19 ts=(nowx-x).total_seconds()
  20 print int(ts)
  21 ### add , subtract days
  22 import datetime
  23 x = datetime.datetime(2014,1,2)
  24 print x
  25 y = x - datetime.timedelta(days=1)
  26 print y
  27 z = y - datetime.timedelta(days=1)
  28 print z
  29 
  30 ### convert string with date to time tuple
  31 import time
  32 timeTuple = time.strptime('2014-01-01 10:11:12','%Y-%m-%d %H:%M:%S')
  33 nrSecondsSince1970=time.mktime(timeTuple)
  34 ### get seconds since epoch
  35 start = int(datetime.datetime(2014,3,9,0,0,0,0).strftime("%s"))*1000
  36 end = int(datetime.datetime(2014,3,14,23,59,59,999).strftime("%s"))*1000
  37 
  38 ### string from utc current date
  39 currutc=datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")

Write and reading data for a plist file

A plist file stores data in XML format.

   1 import plistlib
   2 value = [{'key1':123,'key2':'asdf'},{'keyx1':'testing','keyz1':'yup'}]
   3 # save value in plist file
   4 plistlib.writePlist(value,'/tmp/plist1.plist')
   5 o=plistlib.readPlist('/tmp/plist1.plist')
   6 print o

Content of the file /tmp/plist1.plist

   1 <?xml version="1.0" encoding="UTF-8"?>
   2 <!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
   3 <plist version="1.0">
   4 <array>
   5         <dict>
   6                 <key>key1</key>
   7                 <integer>123</integer>
   8                 <key>key2</key>
   9                 <string>asdf</string>
  10         </dict>
  11         <dict>
  12                 <key>keyx1</key>
  13                 <string>testing</string>
  14                 <key>keyz1</key>
  15                 <string>yup</string>
  16         </dict>
  17 </array>
  18 </plist>

Threading

   1 #!/usr/bin/python
   2 # timestable.py
   3 # calculates the times table in concurrency
   4 import threading
   5 import time
   6 
   7 class TimesTable (threading.Thread):
   8     def __init__(self, timesTable):
   9         threading.Thread.__init__(self) #required
  10         self.timeTable = timesTable
  11         self.count = 1
  12     def run(self):
  13         loop=True
  14         while loop:
  15             time.sleep(1) #sleep for 1 second
  16             result=self.timeTable * self.count
  17             print "%d*%d=%d"%(self.timeTable,self.count,result)
  18             if self.count<10:
  19                 self.count = self.count+1
  20             else:
  21                 self.count=1
  22 
  23 # create threads
  24 timesTable2 = TimesTable(2)
  25 timesTable5 = TimesTable(7)
  26 
  27 # start the threads
  28 timesTable2.start()
  29 timesTable5.start()

unit tests (pyunit)

   1 #filename: unitTest.py
   2 #Run tests: python unitTest.py
   3 #Should fail on testAdd2
   4 import unittest
   5 
   6 class SimpleTestCase(unittest.TestCase):
   7     def setUp(self):
   8         self.res=4
   9 
  10     def tearDown(self):
  11         self.res=0
  12 
  13     def testAdd1(self):
  14         res=2+2
  15         self.assertEqual(self.res,res,'')       
  16 
  17     def testAdd2(self):
  18         res=2+3
  19         self.assertEqual(self.res,res,'')       
  20 
  21     def testAdd3(self):
  22         res=2+3
  23         self.assertNotEqual(self.res,res,'')       
  24 
  25 if __name__ == '__main__':
  26     unittest.main()

unit tests coverage (pyunit)

http://coverage.readthedocs.io/en/latest/cmd.html
pip install coverage
pip install --user --upgrade coverage

tests.py

   1 import unittest
   2 from impl import add 
   3 from impl import sub
   4 
   5 class TestCase(unittest.TestCase):
   6     def test_add(self):
   7         self.assertEqual(5,add(2,3))
   8 #    def test_sub(self):
   9 #        self.assertEqual(1,add(3,2))
  10 if __name__ == '__main__':
  11     unittest.main()

impl.py

   1 def add(arg1,arg2):
   2     return arg1+arg2
   3 
   4 def sub(arg1,arg2):
   5     return arg-+arg2

Steps:

coverage erase
coverage run tests.py
coverage report -m impl.py
coverage html impl.py #in htmlcov folder

cython

Installation:

su
cd /tmp
wget http://cython.org/release/Cython-0.19.1.tar.gz
tar xvzf Cython-0.19.1.tar.gz
cd Cython-0.19.1
python setup.py build
python setup.py install

pymssql

Requires cython. Installation:

su
cd /tmp
wget http://pymssql.googlecode.com/files/pymssql-2.0.0b1-dev-20111019.tar.gz
tar xvvzf pymssql-2.0.0b1-dev-20111019.tar.gz
cd pymssql-2.0.0b1-dev-20111019
python setup.py build
python setup.py install
python
import pymssql

pywhois

Python module for retrieving WHOIS information of domains http://code.google.com/p/pywhois/.

Fetch code with mercurial:

cd /tmp
hg clone https://code.google.com/p/pywhois/
cd pywhois
python setup.py build
python setup.py install

Syntax highlighting on Vim for wsgi

Edit ~./.vimrc:

   1 syntax on
   2 filetype on
   3 au BufNewFile,BufRead *.wsgi set filetype=python

Get file modification time

   1 import os
   2 nrSeconds=os.path.getmtime('/folder/file') # unix epoch, nr secs since 1970-01-01 00:00:00

Kill process with PID

   1 import os
   2 import signal
   3 os.kill(pid,signal.SIGTERM)

Log message to /var/log/messages or /var/log/syslog

   1 import syslog
   2 syslog.syslog(syslog.LOG_INFO,'message to syslog')

   1 import syslog
   2 
   3 syslog.syslog(syslog.LOG_DEBUG,'DEBUG')
   4 syslog.syslog(syslog.LOG_INFO,'INFO')
   5 syslog.syslog(syslog.LOG_WARNING,'WARNING')
   6 syslog.syslog(syslog.LOG_ERR,'ERR')
   7 syslog.syslog(syslog.LOG_CRIT,'CRIT')

OS	File	Logged levels
Slack64 14	/var/log/messages	INFO
Slack64 14	/var/log/syslog	WARNING ERR CRIT
CentOS 6.4	/var/log/messages	INFO WARNING ERR CRIT
Ubuntu 12.04 Precise	/var/log/syslog	DEBUG INFO WARNING ERR CRIT
Debian 7.0 Wheezy	/var/log/syslog	DEBUG INFO WARNING ERR CRIT

Simple process monitor

   1 # example service, dummyService.py
   2 import time
   3 while True:
   4     try:
   5         f=open('filex.log','wb')
   6         f.write('%f'%(time.time()))
   7         f.close() 
   8         time.sleep(5)
   9     except KeyboardInterrupt:
  10         quit()

   1 #service monitor.py
   2 # * * * * * /usr/bin/python /home/vitor/svc/monitor.py
   3 import os
   4 import syslog
   5 import subprocess
   6 
   7 if __name__=="__main__":
   8     files = os.listdir('/proc')
   9     script="dummyService.py"
  10     prefix="dummyService:"
  11 
  12     svcRunning=False
  13 
  14     for file in files:
  15 
  16         if file.isdigit():
  17             cmdline = open('/proc/%s/cmdline'%(file),'r').read()
  18             proc = "%s %s "%(file,cmdline)
  19 
  20             if script  in proc :
  21                 svcRunning=True
  22 
  23     if svcRunning==False:
  24         syslog.syslog(syslog.LOG_INFO,'%s process created '%(prefix) )
  25         # run exec process on python Popen
  26         subprocess.Popen(['/usr/bin/python','dummyService.py'],cwd='/home/userx/svc')

Read URL content

   1 import urllib2
   2 r=urllib2.urlopen('http://www.sapo.pt')
   3 resp=r.read()
   4 print resp

Windows service

Install http://sourceforge.net/projects/pywin32/files/pywin32/Build%20218/pywin32-218.win32-py2.5.exe/download for Python 2.5 for Windows.

Adapted from http://ryrobes.com/python/running-python-scripts-as-a-windows-service/

   1 # C:\cmt\test>python testWinSvc.py install
   2 # C:\cmt\test>python testWinSvc.py start
   3 # C:\cmt\test>python testWinSvc.py stop
   4 # C:\cmt\test>python testWinSvc.py remove
   5 # C:\cmt\test>service.msc
   6 import win32service
   7 import win32serviceutil
   8 import win32api
   9 import win32con
  10 import win32event
  11 import win32evtlogutil
  12 import os, sys, string, time
  13 
  14 class aservice(win32serviceutil.ServiceFramework):
  15    _svc_name_ = "MyServiceShortName"
  16    _svc_display_name_ = "My Service Long Fancy Name!"
  17    _svc_description_ = "THis is what my crazy little service does - aka a DESCRIPTION! WHoa!"
  18          
  19    def __init__(self, args):
  20            win32serviceutil.ServiceFramework.__init__(self, args)
  21            self.hWaitStop = win32event.CreateEvent(None, 0, 0, None)           
  22 
  23    def SvcStop(self):
  24            self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING)
  25            win32event.SetEvent(self.hWaitStop)                    
  26          
  27    def SvcDoRun(self):
  28       import servicemanager      
  29       servicemanager.LogMsg(servicemanager.EVENTLOG_INFORMATION_TYPE,servicemanager.PYS_SERVICE_STARTED,(self._svc_name_, ''))       
  30       self.timeout = 1000     #1 second
  31       # This is how long the service will wait to run / refresh itself (see script below)
  32 
  33       while 1:
  34          # Wait for service stop signal, if I timeout, loop again
  35          rc = win32event.WaitForSingleObject(self.hWaitStop, self.timeout)
  36          # Check to see if self.hWaitStop happened
  37          if rc == win32event.WAIT_OBJECT_0:
  38             # Stop signal encountered
  39             servicemanager.LogInfoMsg("SomeShortNameVersion - STOPPED!")  #For Event Log
  40             break
  41          else:
  42                  try:
  43                      handle = open('c:/windows/temp/outx1.txt', "a")
  44                      handle.write('%s \n' % ('tst svc'))
  45                      handle.close()
  46                  except:
  47                      pass
  48 
  49 
  50 
  51 def ctrlHandler(ctrlType):
  52    return True
  53                   
  54 if __name__ == '__main__':   
  55    win32api.SetConsoleCtrlHandler(ctrlHandler, True)   
  56    win32serviceutil.HandleCommandLine(aservice)

Logging

Adapted from http://docs.python.org/2/howto/logging-cookbook.html#logging-to-multiple-destinations

   1 import logging
   2 
   3 logging.basicConfig(level=logging.DEBUG, \
   4     format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s', \
   5     datefmt='%Y-%m-%d %H:%M:%S %z', \
   6     filename='/tmp/myapp.log', \
   7     filemode='a')
   8 #log to console
   9 console = logging.StreamHandler()
  10 console.setLevel(logging.INFO)
  11 formatter = logging.Formatter('%(name)-12s: %(levelname)-8s %(message)s')
  12 console.setFormatter(formatter)
  13 logging.getLogger('').addHandler(console)
  14 logging.info('Jackdaws love my big sphinx of quartz.') # using default logger '' 
  15 #other loggers
  16 logger1 = logging.getLogger('myapp.area1')
  17 logger2 = logging.getLogger(__name__)
  18 logger1.debug('Quick zephyrs blow, vexing daft Jim.')
  19 logger1.info('How quickly daft jumping zebras vex.')
  20 logger2.warning('Jail zesty vixen who grabbed pay from quack.')
  21 logger2.error('The five boxing wizards jump quickly.')

Copied from http://docs.python.org/2/howto/logging-cookbook.html#configuration-server-example Logging configurable on runtime, receiving new parameters by socket

Server:

   1 import logging
   2 import logging.config
   3 import time
   4 import os
   5 
   6 # read initial config file
   7 logging.config.fileConfig('logging.conf')
   8 
   9 # create and start listener on port 9999
  10 t = logging.config.listen(9999)
  11 t.start()
  12 
  13 logger = logging.getLogger('simpleExample')
  14 
  15 try:
  16     # loop through logging calls to see the difference
  17     # new configurations make, until Ctrl+C is pressed
  18     while True:
  19         logger.debug('debug message')
  20         logger.info('info message')
  21         logger.warn('warn message')
  22         logger.error('error message')
  23         logger.critical('critical message')
  24         time.sleep(5)
  25 except KeyboardInterrupt:
  26     # cleanup
  27     logging.config.stopListening()
  28     t.join()

Client:

   1 #!/usr/bin/env python
   2 import socket, sys, struct
   3 
   4 with open(sys.argv[1], 'rb') as f:
   5     data_to_send = f.read()
   6 
   7 HOST = 'localhost'
   8 PORT = 9999
   9 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  10 print('connecting...')
  11 s.connect((HOST, PORT))
  12 print('sending config...')
  13 s.send(struct.pack('>L', len(data_to_send)))
  14 s.send(data_to_send)
  15 s.close()
  16 print('complete')

Logging with date rotation and email

   1 import logging
   2 from logging.handlers import TimedRotatingFileHandler
   3 from logging.handlers import SMTPHandler
   4 
   5 class Logger(object):
   6     def __init__(self):
   7         loggerName='appx'
   8         logFormatter = logging.Formatter('[%(asctime)s %(name)-12s %(levelname)-8s] %(message)s',datefmt='%Y-%m-%d %H:%M:%S %z')
   9         
  10         rotatingLogHandler = TimedRotatingFileHandler('%s.log'%(loggerName),when='midnight',utc=True)
  11         rotatingLogHandler.setFormatter(logFormatter)
  12         rotatingLogHandler.setLevel(logging.DEBUG)
  13 
  14         smtpLogHandler = SMTPHandler(mailhost='hostx.com', fromaddr='mail@hostx.com', toaddrs='dest@hostx.com', subject=loggerName, \
  15             credentials=('accountx','12345678'), secure=None)
  16         smtpLogHandler.setFormatter(logFormatter)
  17         smtpLogHandler.setLevel(logging.WARNING)
  18                         
  19         self.logger = logging.getLogger(loggerName)
  20         self.logger.addHandler(rotatingLogHandler)
  21         self.logger.addHandler(smtpLogHandler)
  22         self.logger.setLevel(logging.DEBUG)
  23         
  24     def debug(self,m):
  25         self.logger.debug(m)
  26 
  27     def info(self,m):
  28         self.logger.info(m)
  29 
  30     def warning(self,m):
  31         self.logger.warning(m)
  32         
  33     def error(self,m):
  34         self.logger.error(m)
  35         
  36     def critical(self,m):
  37         self.logger.critical(m)

pytz - World Timezone Definitions for Python

   1 import pytz
   2 import datetime
   3 
   4 #fmt = '%Y-%m-%d %H:%M:%S %Z%z'
   5 london=pytz.timezone('Europe/London')
   6 lisbon=pytz.timezone('Europe/Lisbon')
   7 paris  =pytz.timezone('Europe/Paris')
   8 utc = pytz.timezone('UTC')
   9 berlin = pytz.timezone('Europe/Berlin')
  10 
  11 print
  12 dtx = datetime.datetime(2002, 12, 27, 12, 0, 0,tzinfo=utc ) 
  13 print 
  14 print 'UTC    ' , dtx
  15 print 'London ',dtx.astimezone(london)
  16 print 'Lisbon ',dtx.astimezone(lisbon)
  17 print 'Paris  ',dtx.astimezone(paris)
  18 print 'Berlin ',dtx.astimezone(berlin)
  19 
  20 
  21 print
  22 dty = datetime.datetime(2002, 8, 27, 13, 0, 0,tzinfo=utc ) 
  23 print 'UTC    ',dty
  24 print 'London ',dty.astimezone(london)
  25 print 'Lisbon ',dty.astimezone(lisbon)
  26 print 'Paris  ',dty.astimezone(paris)
  27 print 'Berlin ',dty.astimezone(berlin)

Sort example

   1 class Acme(object):
   2     def __init__(self,valuex):
   3         self.valuex=valuex
   4 
   5     def __cmp__(self, other):
   6         if self.valuex==other.valuex:
   7             return 0
   8         if self.valuex>other.valuex:
   9             return 1
  10         if self.valuex<other.valuex:
  11             return -1
  12 
  13     def __repr__(self):
  14         return self.valuex
  15 
  16 if __name__=='__main__':
  17     listx=[]
  18     listx.append( Acme('C') )
  19     listx.append( Acme('A') )
  20     listx.append( Acme('Z') )
  21     print 'original:',listx
  22     print 'sorted asc:',sorted(listx,reverse=False)
  23     print 'sorted desc:',sorted(listx,reverse=True)
  24     listx.sort()
  25     print 'sort:',listx
  26     listx.reverse()
  27     print 'reverse:',listx

sys.path

With sys.path we are able to add new folders which serve as base path to modules. The idea is similar to classpath in Java.

Create file /tmp/testex.py

   1 def test1()
   2     print("Alright !!!!")

Run python interactively:

   1 vitor@darkstar[~]$python
   2 Python 2.7.3 (default, Jul  3 2012, 21:16:07) 
   3 [GCC 4.7.1] on linux2
   4 Type "help", "copyright", "credits" or "license" for more information.
   5 >>> import testex
   6 Traceback (most recent call last):
   7   File "<stdin>", line 1, in <module>
   8 ImportError: No module named testex
   9 >>> import sys
  10 >>> sys.path.append('/tmp') # add /tmp to modules paths
  11 >>> testex.test1()
  12 Allright !!!!

By default on Linux systems, modules are also searched in ~/.local/pythonXX.YY/site-packages. If the file testex.py would exist, it would also be used without being necessary to add the folder /tmp to the sys.path . For the user vitor and python 2.7 the path would be /home/vitor/.local/lib/python2.7/site-packages .

Math

   1 # round float to 6 decimal places
   2 nr=1.23456789
   3 print round(nr,6)
   4 # round float to 4 decimal places
   5 nr=1.23456789
   6 print round(nr,4)

SMTP

   1 import smtplib
   2 
   3 class Smtp(object):
   4     def __init__(self,server,port,user,passx,appName):
   5         self.server=server
   6         self.port=port
   7         self.user=user
   8         self.passx=passx
   9         self.appName=appName
  10 
  11     def send(self,fromx,tox,subjectx,contentType,body):
  12         s=smtplib.SMTP()
  13         s.connect(self.server,self.port)
  14         s.ehlo(self.appName)
  15         s.login(self.user,self.passx)         
  16         msg='From: %s\r\nTo: %s\r\nSubject: %s\r\nContent-Type: %s\r\n\r\n%s'%(fromx,tox,subjectx,contentType,body)
  17         s.sendmail(fromx,tox,msg)
  18         s.quit()
  19 
  20 s=Smtp('mail.example.org',25,'norep@example.org','12345678','TestApp')
  21 # content types: text/plain text/html
  22 s.send('norep@example.org',['user1@example.org','user2@example.org'],'Subject','text/plain','Body mail')

JSON serializable

   1 from json import JSONEncoder
   2 import json
   3 
   4 class EncoderJSON(JSONEncoder):
   5     def default(self, o):
   6         return o.getJSONDict()    
   7     
   8 class Acme(object):
   9     def __init__(self,paramx,valuex):
  10         self.paramx=paramx
  11         self.valuex=valuex
  12         
  13     def getJSONDict(self):
  14         return {'paramx':self.paramx ,'valuex':self.valuex}
  15                 
  16     def __repr__(self):
  17         return '%s %s %s'%(self.paramx ,self.valuex)
  18 
  19 if __name__=='__main__':
  20     listx=[]
  21     listx.append(Acme('p1','v1'))
  22     listx.append(Acme('p2','v2'))
  23     # json.dumps likes lists and dictionaries
  24     print( json.dumps(listx,cls=EncoderJSON) ) 
  25     # output [{"valuex": "v1", "paramx": "p1"}, {"valuex": "v2", "paramx": "p2"}]

Decorators

http://simeonfranklin.com/blog/2012/jul/1/python-decorators-in-12-steps/

Wraps functions with functionality, before, after , changes parameters of the decorated function.

   1 #decorator logger
   2 def logger(func):
   3         def inner(*args, **kwargs): # step 1
   4             print "Arguments were: %s, %s" % (args, kwargs)
   5             return func(*args, **kwargs) # step 2
   6         return inner
   7 
   8 @logger
   9 def foo1(x, y=1):
  10     return x * y
  11 if __name__=='__main__':
  12     foo1(2,2)

   1 import sys
   2 from os import path 
   3 #decorator adder
   4 def add_one_args(decoratedFunction):
   5     def interceptor(*args, **kwargs): # step 1
   6         print "Arguments were: args %s, kwargs %s" % (args, kwargs)
   7         index=0
   8         new_args=[]
   9         # change input 
  10         isnotint=False
  11         for arg in args:
  12             if isinstance(arg,int):                    
  13                 new_args.append(arg+1)
  14             else:
  15                 isnotint=True
  16             index=index+1
  17                 
  18         if isnotint==False: 
  19             args=tuple(new_args)
  20 
  21         new_kwargs={}
  22         for kwarg in kwargs:
  23             new_kwargs[kwarg] = kwargs[kwarg]+1
  24 
  25         decoratedFuncReturn =  decoratedFunction(*args, **new_kwargs) # step 2
  26         # change output
  27         decoratedFuncReturn = decoratedFuncReturn * 2
  28         return decoratedFuncReturn
  29     return interceptor
  30 
  31 @add_one_args
  32 def foo1(x, y=1):
  33     print('Got arguments in foo1 %d %d'%(x,y))
  34     mul = x * y
  35     print('Multiplication result inside function %d'%(mul))
  36     return mul
  37 
  38 def foo2(**kwargs):
  39     print('Got arguments in foo1 %d %d'%( kwargs['x'] , kwargs['y']  ))
  40     mul = kwargs['x'] * kwargs['y']
  41     print('Multiplication result inside function %d'%(mul))
  42     return mul
  43 
  44 if __name__=='__main__':
  45     print( foo1(2,y=2) )
  46     print( foo2( x=3,y=3  ) )
  47     print( sys.path )
  48     print( sys.prefix )
  49     print( sys.exec_prefix )
  50     here = path.abspath(path.dirname(__file__))
  51     print(here)

   1 def add_one_args(decoratedFunction,xyz=1):
   2     def interceptor(*args, **kwargs):
   3         print "Arguments were: args %s, kwargs %s" % (args, kwargs)
   4         new_args=[]
   5 
   6         for arg in args:
   7             if isinstance(arg,int): new_args.append(arg+1)
   8             else: new_args.append(arg)
   9 
  10         args=tuple(new_args)
  11         decoratedFuncReturn = decoratedFunction(*args, **kwargs) 
  12         return decoratedFuncReturn
  13     return interceptor
  14 
  15 def subtract_one_args(decoratedFunction,xyz=1):
  16     def interceptor(*args, **kwargs):
  17         print "Arguments were: args %s, kwargs %s" % (args, kwargs)
  18         new_args=[]
  19 
  20         for arg in args:
  21             if isinstance(arg,int): new_args.append(arg-1)
  22             else: new_args.append(arg)
  23 
  24         args=tuple(new_args)
  25         decoratedFuncReturn = decoratedFunction(*args, **kwargs) 
  26         return decoratedFuncReturn
  27     return interceptor
  28 
  29 class Acme(object):
  30     def __init__(self):
  31         self.a = 0
  32         self.b = 0
  33 
  34     @add_one_args
  35     @subtract_one_args
  36     def set(self,a,b):
  37         self.a=a
  38         self.b=b
  39         print self.a, self.b
  40 
  41 if __name__=='__main__':
  42     x = Acme()
  43     x.set(0,1)

Dictionaries

   1 a={} #create dictionary
   2 if 'keyx' in a: print 'ok' #check if key keyx exists
   3 a['keyx']=1 # create key keyx with value 1
   4 if 'keyx' in a: print 'ok' #checks  keyx existence
   5 a['key2']={} # create key key2
   6 if 'keyx' in a['key2']: print 'ok' # check if keyx exists in key2
   7 a['key2']['keyx']=3 # create keyx inside key2
   8 if 'keyx' in a['key2']: print 'ok'

Exceptions

Example 1

   1 def dummyMethod(datax):
   2     try:
   3         if 'key1' in datax:
   4             key1 = datax['key1']
   5 
   6             if 'key2' in key1:
   7                 key2 = dvi['key2']
   8                 print('key2 is %s '%(key2))                    
   9 
  10     except Exception,ex:
  11         print('Error in dummyMethod %s'%( str(ex) ) )

Install pip

easy_install pip

Regular expressions

   1 import re
   2 listx=['aa12.txt','a123sss.txt','bs11bb.txt','123aaaa.sql','a12.txt','aaaa12','20ghj','6657']
   3 
   4 for item in listx:
   5     res = re.match('^(\D+)(\d+)(\D+)$',item)
   6     if res!=None:
   7         decimal =  int(res.group(2)) + 1
   8         print('A: mv %s %s%d%s'%(item , res.group(1),decimal , res.group(3) ))
   9     res = re.match('^(\d+)(\D+)$',item)
  10     if res!=None:
  11         decimal =  int(res.group(1)) + 1
  12         print('B: mv %s %d%s'%(item , decimal,res.group(2)))
  13     res = re.match('^(\D+)(\d+)$',item)
  14     if res!=None:
  15         decimal =  int(res.group(2)) + 1
  16         print('C: mv %s  %s%d'%(item , res.group(1),decimal))
  17     res = re.match('^(\d+)$',item)
  18     if res!=None:
  19         decimal =  int(res.group(1)) + 1
  20         print('D: mv %s  %d'%(item , decimal))
  21 '''
  22 #ouput
  23  python addNumber.py 
  24 A: mv aa12.txt aa13.txt
  25 A: mv a123sss.txt a124sss.txt
  26 A: mv bs11bb.txt bs12bb.txt
  27 B: mv 123aaaa.sql 124aaaa.sql
  28 A: mv a12.txt a13.txt
  29 C: mv aaaa12  aaaa13
  30 B: mv 20ghj 21ghj
  31 D: mv 6657  6658
  32 '''

Command line arguments

import sys

   1 import sys
   2 # ...
   3 if __name__=='__main__':
   4     intArgA = int(sys.argv[1])
   5     intArgB = int(sys.argv[2])
   6     intArgC = int(sys.argv[3])

Run python code in shell without source code file .py

   1 echo "<log type="typex" other="bbbbb" msgId="bbbbbb" >" | xargs -i echo -e "strx='{}'\nsplitted = strx.split(' ')\nfor spl in splitted:\n\tif 'msgId' in spl:\n\t\tprint spl.split('=')[1]"  | python

   1 strx='{}' # where xargs value is used
   2 splitted = strx.split(' ')
   3 for spl in splitted:
   4     if 'msgId' in spl:
   5         print spl.split('=')[1]

Alias identifier inside module

>>> from json import dumps as jsonDumps
>>> jsonDumps({'key':'value'})
'{"key": "value"}'
>>> print jsonDumps({'key':'value' , 'key2':1234})
{"key2": 1234, "key": "value"}
>>>

Execute process using subprocess

   1 import subprocess 
   2 
   3 if __name__=='__main__':
   4   try:
   5     args=[]
   6     args.append('ls')
   7     args.append('/tmpx')
   8     out = subprocess.check_output(args, stderr=subprocess.STDOUT, shell=False)
   9     print( out )
  10   except subprocess.CalledProcessError as ex1:
  11     print("Exception command: %s"%(ex1.cmd)  )    
  12     print("Exception code: %s"%(ex1.returncode)  )    
  13     print("Exception output: %s"%(ex1.output)  )
  14   except Exception as ex2:
  15     print("General exception %s"%(ex2.message) )

Recursion

   1 def factorial(n):
   2   if n > 0: return n * factorial(n-1)
   3   else: return 1
   4 
   5 def fibo(n):
   6   if n==0: return 0
   7   if n==1: return 1
   8   if n>1: return fibo(n-1) + fibo(n-2)
   9 
  10 def sum(idx,lista):
  11   if idx<0: return 0
  12   if idx>=0: return lista[idx] + sum(idx-1,lista)
  13 
  14 if __name__=='__main__':
  15   print( factorial(4) )
  16   print( fibo(3) )
  17   lst=[1,2,3,4]
  18   print( sum(len(lst)-1,lst) )

Python3 virtual environment

https://packaging.python.org/tutorials/installing-packages/#creating-virtual-environments

Python “Virtual Environments” allow Python packages to be installed in an isolated location for a particular application, rather than being installed globally.

You can deactivate a virtual environment by typing “deactivate” in your shell.

   1 sudo apt-get install python3-venv
   2 python3 -m venv testVenv
   3 source testVenv/bin/activate
   4 pip install --upgrade pip
   5 pip install -i https://testpypi.python.org/pypi vbhelloworld
   6 pip uninstall vbhelloworld

https://docs.python.org/3/library/venv.html#module-venv

When a virtual environment is active (i.e., the virtual environment’s Python interpreter is running), the attributes sys.prefix and sys.exec_prefix point to the base directory of the virtual environment, whereas sys.base_prefix and sys.base_exec_prefix point to the non-virtual environment Python installation which was used to create the virtual environment.

If a virtual environment is not active, then sys.prefix is the same as sys.base_prefix and sys.exec_prefix is the same as sys.base_exec_prefix (they all point to a non-virtual environment Python installation).

List Comprehensions

   1 squares = [x**2 for x in range(10)]
   2 # squares = [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
   3 sequence = [x for x in range(1,9)]
   4 # sequence [1, 2, 3, 4, 5, 6, 7, 8]
   5 range(10)
   6 # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
   7 squares = [x**2 for x in range(10) if x<= 4]
   8 # squares [0, 1, 4, 9, 16]

Python 2 virtual env

   1 # in windows with powershell
   2 cd ~
   3 pip install virtualenv --user
   4 # add the scripts folder to the user path c:\users\userx\AppData\Roaming\Python\Scripts 
   5 # using cmd
   6 cd %HOME_PATH%
   7 mkdir tmp
   8 cd tmp
   9 virtualenv.exe test_env
  10 cd test_env
  11 Scripts\activate.bat

   1 import sys
   2 sys.path

Generator functions yield

   1 #test yield
   2 def factorial(val):
   3     v=val
   4     res=1
   5     while v>1:
   6         res=res*v
   7         v=v-1
   8         yield None # v
   9     yield res
  10 
  11 if __name__ == "__main__":
  12     for i in factorial(3):
  13         print(i)
  14 
  15     f = factorial(4)
  16     iterate=True
  17 
  18     while iterate:
  19         try:
  20             print( f.next() )
  21             #print( next(f) )
  22         except StopIteration as ex:
  23             iterate=False

Output:

   1 python test_yeld.py 
   2 None
   3 None
   4 6
   5 None
   6 None
   7 None
   8 24

XML parse with minidom

https://docs.python.org/2/library/xml.dom.minidom.html

   1 from xml.dom import minidom
   2 print("Hello world")
   3 a="<a><b>aaaa</b><b>bbcc</b></a>"
   4 doc = minidom.parseString(a) # return document 
   5 tags = doc.getElementsByTagName("b")
   6 for t in tags:
   7   print(t.firstChild.nodeValue)

Environment variable

   1 import os
   2 print(os.environ)

XSD and objectify XML

test.py

   1 from lxml import objectify
   2 from StringIO import StringIO
   3 from lxml import etree
   4 
   5 f = open('test.xsd')
   6 xsddata=''
   7 for line in f:
   8     xsddata = xsddata +  line
   9 f.close()
  10 
  11 xmlschema_doc = etree.parse(StringIO(xsddata))
  12 xmlschema = etree.XMLSchema(xmlschema_doc)
  13 print xmlschema 
  14 
  15 parser = objectify.makeparser(schema=xmlschema)
  16 xml = "<a><b>test</b></a>"
  17 a = objectify.fromstring(xml, parser)
  18 print(a.b)

test.xsd

   1 <xsd:schema xmlns:xsd="http://www.w3.org/2001/XMLSchema">
   2   <xsd:element name="a" type="AType"/>
   3   <xsd:complexType name="AType">
   4        <xsd:sequence>
   5          <xsd:element name="b" type="xsd:string" />
   6        </xsd:sequence>
   7   </xsd:complexType>
   8 </xsd:schema>

Other example ....

@startuml
class AType
class BType
class AcmeType

/' 
<|--  extends 
-- association
..> depends 
<|.. implements / realizes
'/ 

AType : b : String
AType : xx : BType
AcmeType : other : int
BType : c : String

AType ..> BType
AcmeType --|> AType
@enduml

acme.xsd

   1 <xs:schema 
   2     xmlns:xs="http://www.w3.org/2001/XMLSchema" 
   3     xmlns:bit="http://bitarus.allowed.org/types" 
   4     xmlns:ot="http://bitarus.allowed.org/othertypes" 
   5     targetNamespace="http://bitarus.allowed.org/othertypes" 
   6     elementFormDefault="qualified"
   7   >
   8   <xs:import schemaLocation="test.xsd" />  
   9   <xs:complexType name="AcmeType">  
  10     <xs:complexContent>
  11         <xs:extension base="bit:AType">
  12             <xs:sequence>
  13                 <xs:element name="other" type="xs:int"/>
  14             </xs:sequence>
  15         </xs:extension>
  16     </xs:complexContent>      
  17   </xs:complexType>
  18   <xs:element name="at" type="ot:AcmeType"/>
  19 </xs:schema>

test.xsd

   1 <xsd:schema 
   2     xmlns:xsd="http://www.w3.org/2001/XMLSchema" 
   3     xmlns:bit="http://bitarus.allowed.org/types" 
   4     targetNamespace="http://bitarus.allowed.org/types" 
   5     elementFormDefault="qualified" 
   6 >
   7   <xsd:complexType name="AType">
   8        <xsd:sequence>
   9          <xsd:element name="b" type="xsd:string" />         
  10          <xsd:element name="xx" type="bit:BType" />
  11        </xsd:sequence>
  12   </xsd:complexType>  
  13   <xsd:complexType name="BType">
  14       <xsd:sequence>
  15         <xsd:element name="c" type="xsd:string" />
  16       </xsd:sequence>
  17   </xsd:complexType>  
  18   <xsd:element name="a" type="bit:AType"/>
  19 </xsd:schema>

acme.py

   1 from lxml import objectify
   2 from StringIO import StringIO
   3 from lxml import etree
   4 import os
   5 f = open('acme.xsd')
   6 xsddata=''
   7 for line in f:
   8     xsddata = xsddata +  line
   9 f.close()
  10 
  11 xmlschema = etree.XMLSchema(etree.parse(StringIO(xsddata)))
  12 
  13 xml = """
  14 <ot:at xmlns:ot="http://bitarus.allowed.org/othertypes" 
  15   xsi:schemaLocation="http://bitarus.allowed.org/othertypes other.xsd" 
  16   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  17   xmlns:bit="http://bitarus.allowed.org/types" >
  18   <bit:b>cccc</bit:b>
  19   <bit:xx><bit:c>hhjjkk</bit:c></bit:xx>
  20   <ot:other>1234</ot:other>  
  21 </ot:at>
  22 """
  23 
  24 doc = etree.fromstring(xml, etree.XMLParser(schema=xmlschema) )
  25 # https://lxml.de/api/lxml.etree._Element-class.html
  26 bNode = doc.getchildren()[0]
  27 otherNode = doc.getchildren()[2]
  28 print bNode.text
  29 print otherNode.text

XML tree/graph

   1 from lxml import objectify
   2 from lxml import etree
   3 from StringIO import StringIO
   4 import sys
   5 
   6 def show_children(parent,node):
   7     if node.getchildren():
   8         print parent+"/"+node.tag
   9         index=1
  10         for c in node.getchildren():
  11             show_children(parent+"/"+node.tag, c )
  12             index=index+1
  13     else:
  14         if node.text:
  15             print parent + "/"+node.tag + "=" + node.text
  16         else:
  17             print parent + "/"+node.tag
  18 
  19     if node.attrib:
  20         for key in node.attrib:
  21             print parent + "/"+node.tag +"@"+key+"="+node.attrib[key]
  22 
  23 if __name__=='__main__':
  24     f = open(sys.argv[1])
  25     data=''
  26     for line in f:
  27        if line.startswith("<?xml ")==False:  data = data +  line
  28     f.close()
  29     # print data
  30     doc = etree.fromstring( data )
  31     f.close()
  32     show_children("",doc)

Debugger

   1 # python test.py
   2 # DEBUG=on python test.py  
   3 import os
   4 # ....
   5 if 'DEBUG' in os.environ:
   6   # breakpoint 
   7   import pdb; pdb.set_trace()
   8 
   9 # commands
  10 # c continue
  11 # q quit
  12 # n next
  13 # l list
  14 # s step

Remote debugger

   1 # pip install --user rpdb
   2 # cat rpdb_test.py
   3 print("aaa")
   4 # listens in 0.0.0.0 4444
   5 import rpdb; rpdb.set_trace(addr="0.0.0.0")
   6 print("bbb")
   7 print("ccc")
   8 # python3 rpdb_test.py
   9 # docker run (...) -p 4444:4444
  10 # nc localhost 4444

Install package in user home folder pointing to other libs

rpm -ql rpmpackage # check installed RPM package contents
pip install --user --global-option=build_ext --global-option="-L/usr/lib64" package_to_install

Atom syndication

Identify current function

   1 import inspect
   2 print 'In %s'%( inspect.getframeinfo(inspect.currentframe()).function )

Operative system folder path separator

   1 import os
   2 print( os.sep )
   3 help(os)
   4 dir(os)

Install package for user with a more recent version

   1 pip freeze
   2 pip install --user  numpy --upgrade
   3 pip install --user  --upgrade numpy==1.15.1
   4 pip freeze

with statement

https://www.python.org/dev/peps/pep-0343/

with EXPR as VAR:
    BLOCK
which roughly translates into this:

VAR = EXPR
VAR.__enter__()
try:
    BLOCK
finally:
    VAR.__exit__()

The open() method returns a file object. The file object has the following methods that fit in the with statement:

enter(...)
exit(...) Closes the file.

   1 with open('filex') as fx:
   2     pass

reST docstring format

http://www.sphinx-doc.org/en/1.6/domains.html#the-python-domain

.. py:function:: send_message(sender, recipient, message_body, [priority=1])

   Send a message to a recipient

   :param str sender: The person sending the message
   :param str recipient: The recipient of the message
   :param str message_body: The body of the message
   :param priority: The priority of the message, can be a number 1-5
   :type priority: integer or None
   :return: the message id
   :rtype: int
   :raises ValueError: if the message_body exceeds 160 characters
   :raises TypeError: if the message_body is not a basestring

Static method

Adapted from https://stackoverflow.com/questions/735975/static-methods-in-python

   1 class ClassX(object):
   2     @staticmethod
   3     def static(x):
   4         print(x)
   5 
   6 ClassX.static(2) # outputs 2

Singleton

Adapted from https://www.tutorialspoint.com/python_design_patterns/python_design_patterns_singleton.htm

   1 class Singleton:
   2    __instance = None
   3 
   4    @staticmethod 
   5    def getInstance():
   6       """ Static access method. """
   7       if Singleton.__instance == None:
   8          Singleton()
   9       return Singleton.__instance
  10 
  11    def __init__(self):
  12       """ Virtually private constructor. """
  13       if Singleton.__instance != None:
  14          raise Exception("This class is a singleton!")
  15       else:
  16          Singleton.__instance = self
  17 
  18 s = Singleton()

Iterate files from a root folder

   1 #!/usr/bin/python
   2 
   3 import os
   4 for root, dirs, files in os.walk(".", topdown=False):
   5    for name in files:
   6       if name.endswith(".txt"):
   7           print(os.path.join(root, name))

xml pretty print

   1 import lxml
   2 f = open('out.xml')
   3 from lxml import etree
   4 doc = etree.fromstring( f.read() )
   5 out = etree.tostring(doc, pretty_print=True)
   6 f.close()
   7 f = open('out2.xml','wb')
   8 f.write( out ) 
   9 f.flush()
  10 f.close()
  11 quit()

docstring sphynx reST

Info field lists
New in version 0.4.
Inside Python object description directives, reST field lists with these fields are recognized and formatted nicely:
    param, parameter, arg, argument, key, keyword: Description of a parameter.
    type: Type of a parameter. Creates a link if possible.
    raises, raise, except, exception: That (and when) a specific exception is raised.
    var, ivar, cvar: Description of a variable.
    vartype: Type of a variable. Creates a link if possible.
    returns, return: Description of the return value.
    rtype: Return type. Creates a link if possible.

future

https://docs.python.org/2/library/__future__.html

   1 from __future__ import absolute_import # 3.0
   2 from __future__ import division # 3.0
   3 from __future__ import generators #2.3
   4 from __future__ import nested_scopes # 2.2
   5 from __future__ import print_function # 3.0
   6 from __future__ import unicode_literals # 3.0
   7 from __future__ import with_statement # 2.6

Command line argument parser

https://docs.python.org/2/library/argparse.html

Parser for command-line options, arguments and sub-commands

Type checking

   1 assert isinstance(obj,Class)
   2 raise NotImplementedError

Private and protected class members

   1 _member # protected member
   2 __member # private member, name mangling 
   3 member # public member

Nested class inside a function

   1 def a():
   2 
   3     class X(object):
   4         CONST=1
   5         def __init__(self):
   6             self._c = X.CONST
   7             print(self._c)
   8 
   9     x=X()
  10 
  11 if __name__=='__main__':
  12     a()

List comprehension

   1 (Pdb) [ val for val in range(1,10) if val % 2 == 0]
   2 [2, 4, 6, 8]

Change XML values with xpath

   1 from lxml import etree, objectify
   2 
   3 xml_data='<a><b><c x="as">def</c></b></a>'
   4 
   5 doc = etree.fromstring(xml_data)
   6 print "Original"
   7 print etree.tostring(doc)
   8 
   9 elem = doc.xpath('//c', namespaces=doc.nsmap)
  10 elem[0].text = "hhhhh"
  11 
  12 node_attr = doc.xpath('/a/b/c', namespaces=doc.nsmap)
  13 node_attr[0].attrib['x']='other_xyz'
  14 
  15 print etree.tostring(doc)

*args and **kwargs

   1 def x(*args,**kwargs):
   2   print type(args) # tuple
   3   print type(kwargs) # dict
   4   print args[0] # prints 1
   5   print args[1] # prints 2
   6   print kwargs['a'] # prints 1234
   7   print kwargs['b'] # prints 4567
   8 
   9 x(1 ,2, a = 1234, b = 4567)

json indentation, pretty print

   1 import json
   2 
   3 def json_to_string(obj):
   4     return json.dumps(obj,indent=4,sort_keys=True)
   5 
   6 if __name__=='__main__':
   7     print json_to_string({'a':11,'b':'fffgg'})

multiple inheritance - python 2

   1 #!/usr/local/bin/python2
   2 
   3 class A(object):
   4     def init_A(self):
   5         print("Called init_A")
   6         self.a = "aaaa"
   7 
   8     def hello(self):
   9         return "hello"
  10 
  11     def get_c(self):
  12         return self.c
  13 
  14 class C(object):
  15     def __init__(self):
  16         self.c = "cccc"
  17         print("Called init C")
  18 
  19     def world(self):
  20         return "world"
  21 
  22 class B(C, A):
  23     """Multiple inheritance from C and A """
  24     def __init__(self):
  25         super(B,self).__init__() # calls first class from which is inherited
  26         super(B,self).init_A()
  27         print(dir(self))
  28 
  29     def hello(self):
  30         return super(B,self).hello() + " " + super(B,self).world() + " " + self.c + " " + self.a
  31 
  32 if __name__=="__main__":
  33     b = B()
  34     print( b.hello() )

Output

   1 openbsd$ ./inheritance.py
   2 Called init C
   3 Called init_A
   4 ['__class__', '__delattr__', '__dict__', '__doc__', '__format__', '__getattribute__', '__hash__', '__init__', '__module__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', '__weakref__', 'a', 'c', 'get_c', 'hello', 'init_A', 'world']
   5 hello world cccc aaaa

-  ⇤ ← Revision 150 as of 2019-09-07 13:28:04 → 
  Size: 42738
  Editor: localhost
  Comment:
+   ← Revision 151 as of 2019-09-07 13:29:15 → ⇥
  Size: 43155
  Editor: localhost
  Comment:
-Deletions are marked like this.
+Additions are marked like this.
 Line 1575:
+Output
{{{#!highlight bash
openbsd$ ./inheritance.py
Called init C
Called init_A
['__class__', '__delattr__', '__dict__', '__doc__', '__format__', '__getattribute__', '__hash__', '__init__', '__module__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', '__weakref__', 'a', 'c', 'get_c', 'hello', 'init_A', 'world']
hello world cccc aaaa
}}}