Python

Índex

General

Instal·lació / Installation
  • Opcions / Options
    1. Normalment Python s'instal·la fent servir la gestió de paquets de la vostra distribució / Usually installed from package in your distribution
    2. Si us cal una versió específica de Python (per exemple per a fer servir des de tox), us la podeu baixar, compilar i instal·lar / If you need a specific version of Python (e.g. to be use with tox), you can download, compile and install it:
      • Download
      • cd Python...
      • ./configure
      • make
      • sudo make altinstall # do not overwrite already installed version
      • uninstall
        • sudo rm -f /usr/local/bin/{python3*,pip3*,2to3*,pyvenv*,easy_install-3*,idle3*,pydoc3*}
        • sudo rm -f /usr/local/lib/libpython3.6m.a
        • sudo rm -rf /usr/local/lib/python3.6
        • sudo rm -rf /usr/local/lib/pkgconfig/python-3.6*
        • sudo rm -rf /usr/local/lib/pkgconfig/{python3.pc,python-3.6m.pc,python-3.6.pc}
    3. pyenv
      • baixarà, compilarà i instal·larà pyhon a ~/.pyenv/versions/
      • Instal·lació / Installation
        • ...
      • Ús / Usage
        • list installed versions
          • global
            • pyenv global
          • local
            • pyenv local
          • shell
            • pyenv shell
        • install versions
          • pyenv install 3.6.12

Python 3
  • Porting Python 2 Code to Python 3
    1. check good coverage of tests: use coverage.py
      • Django: Integration with coverage
        • pip install coverage
        • coverage run --source='.' --omit='env/*,*/migrations/*' manage.py test
        • coverage report
        • coverage html
          • firefox htmlcov/index.html
    2. make your code compatible with python 2 and python 3: use futurize (based on lib2to3 and use fixers from 2to3, 3to2, and python-modernize) or modernize
      • pip install future
      • Django
        • cd myproject
        • stage 1
          • futurize --stage1 myproject/*.py
          • futurize --stage1 -w myproject/*.py
          • for every app:
            • futurize --stage1 myapp/*.py
            • futurize --stage1 -w myapp/*.py
            • futurize --stage1 myapp/*/*.py
            • futurize --stage1 -w myapp/*/*.py
          • run tests
        • stage 2
          • NOTE: if you already made the effort to protect your urllib imports with six.PY2, you may want futurize not to replace imports under six.PY2:
            • futurize --stage2 --nofix=libfuturize.fixes.fix_future_standard_library --nofix=libfuturize.fixes.fix_future_standard_library_urllib myproject/*.py
          • futurize --stage2 myproject/*.py
          • futurize --stage2 -w myproject/*.py
          • for every app:
            • futurize --stage2 myapp/*.py
            • futurize --stage2 -w myapp/*.py
            • futurize --stage2 myapp/*/*.py
            • futurize --stage2 -w myapp/*/*.py
          • run tests
      • ...
    3. clean your code: use pylint
    4. check whether dependent modules can be ported to python 3: use caniusepython3
    5. test your code under several versions of python: use tox
  • Supporting Python 3: An in-depth guide
  • Detection of python version
    • ...
    • using six:
  • Porting Code to Python 3 with 2to3
  • Cheat Sheet: Writing Python 2-3 compatible code
    • Python 2 Python 3
      print print()
      .iteritems() .items()
      class Meta from builtins import object
      class Meta(object)


      string.find()
      		 str.find()
      ....values()
      		 list(....values())
      ....keys() list(....keys())
      ....items() list(....items())
      string.letters string.ascii_letters
      open(my_binary_file) open(my_binary_file, 'rb')
      a.next() next(a)
  • Generador / Generator
  • Strings and bytes
    • Unicode
    • Text Vs. Data Instead Of Unicode Vs. 8-bit

      		• python 2 python 3


        		type creation
        (not recommended)        		 creation
        (recommended)
        (to be used by 2to3)
        		type creation
        unencoded unicode strings <type 'unicode'> u'ànima' unicode(...) text (unicode)
        • <class 'str'> (unmutable)
        		 'ànima'
        encoded, binary 8-bit strings <type 'str'>
        		str('ànima') binary data b'...'
      • conversion in Python3
        • from (row) -> to (column)

          		str bytes
          str
          • my_str.encode()
          • bytes(my_str, encode='utf-8')
          bytes
          • my_bytes.decode()
          • str(my_bytes, encode='utf-8')

Python
  • The Pyhton tutorial
  • PEP
  • Estil / Style
  • Fitxer / File (shebang)
    • #!/usr/bin/env python3
      # -*- coding: utf-8 -*-
    • ...
  • Python (Raspberry Pi)
  • PyPI: the Python package index
  • Python Documentation contents
    • The Python standard library
      • print
        • from __future__ import print_function
      • type
        • if type(myvar) is int
          ...
      • isinstance
        • import six
          if isinstance(myvar, (basestring))
          ...
      • Dades / Data
        • ...
          type create create element set element check for key retrieve element get index remove element join
          Object class MyModel:
              myfield1 = 'myvalue1'

          myobject = MyModel()
          		myobject.myfield1 = 'mynewvalue' hasattr(myobject, 'myfield1')
          • myobject.myfield1
          • getattr(myobject, 'myfield1', 'mydefaultvalue')


          dict
          • mydict = {}
          • mydict = {'mykey1':'myvalue1', 'mykey2':'myvalue2'}
          		 mydict['mykey3'] = 'myvalue3' mydict['mykey1'] = 'mynewvalue1' 'mykey1' in mydict
          • mydict['mykey1']
          • mydict.get('mykey1','mydefaultvalue')
          • mydict.pop('mykey1')
          		 mydict.update(myotherdict)
          list
          • mylist = []
          • mylist = ['myvalue1','myvalue2']
          		 mylist.append('myvalue3')
          		'myvalue1' in mylist
          • mylist[0]
          		 mylist.index('mykey1')
          • mylist.pop(0)
          		 mylist + myotherlist
          tuple
          • mytuple = ()
          • mytuple = ('myvalue1','myvalue2',)
          		 mytuple += ('myvalue3',)
          		'myvalue1' in mytuple
          • mytuple[0]
          		 mylist.index('mykey1')
          		mytuple + myothertuple
        • Create a dict from a list
          • Python : How to convert a list to dictionary ?
          • >>> mylist = [{'first_key':'value_f1', 'second_key':'value_s1'}, {'first_key':'value_f2', 'second_key':'value_s2'}]
            >>> a = {b['first_key']:b['second_key'] for b in mylist}
            >>> a
            {'value_f1': 'value_s1', 'value_f2': 'value_s2'}
        • Search/filter from a list of dicts:
          • a = [{'name':'myname1','address':'myaddress11'}, {'name':'myname1','address':'myaddress12'}, {'name':'myname2','address':'myaddress21'},]
            elements_from_myname1 = [c for c in a if c['name']=='myname1']
            addresses_from_myname1 = [c['address'] for c in a if c['name']=='myname1']
      • Test unitari / Unit test

        		• pytest unittest Django tests (based on unittest) Django drf
          file mytest.py mytest.py myproject/myapp/tests/mytest.py myproject/myapp/tests/mytest.py
          import import pytest from unittest import TestCase from django.test import TestCase from rest_framework.test import APITestCase
          class class MyGroupTests: class MyGroupTestCase(TestCase): class MyGroupTestCase(TestCase):
              """ Tests over database """          		 class MyGroupAPITestCase(APITestCase):
              """ Tests over API REST """
          funció que s'executa sempre en començar def setup_method(self): def setUp(self): def setUp(self): def setUp(self):
          test def test_first(self): def test_first(self): def test_first(self): def test_first(self):
          funció que s'executa sempre en acabar def teardown_method(self): def ...(self)

          run
          • pytest -s mytest.py -k test_first
          • python -m unittest --verbose mytest
          • python -m unittest --verbose mytest.MyGroupTestCase
          • python -m unittest --verbose mytest.MyGroupTestCase.test_first
          • ./manage.py test --keepdb --settings myapp.tests_settings --verbosity 3 myapp.tests.mytest
          • ./manage.py test --keepdb --settings myapp.tests_settings --verbosity 3 myapp.tests.mytest.MyGroupTestCase
          • ./manage.py test --keepdb --settings myapp.tests_settings --verbosity 3 myapp.tests.mytest.MyGroupTestCase.test_first
          • ./manage.py test --keepdb --settings myapp.tests_settings --verbosity 3 my_app.tests.mytest
          • ./manage.py test --keepdb --settings myapp.tests_settings --verbosity 3 my_app.tests.mytest.MyGroupAPITestCase
          • ./manage.py test --keepdb --settings myapp.tests_settings --verbosity 3 my_app.tests.mytest.MyGroupAPITestCase.test_first
          run from Eclipse / PyDev
          • select class or method: Debug as -> Python unit-test

        • pytest
        • unittest
          • Eclipse / PyDev
            • first time
              • selected file: Debug as -> Python unit-test
            • Debug Configurations
              • Arguments
                • Override PyUnit preferences for this launch?
                  • PyDev test runner
                  • --verbosity 3
            • ...
        • Django tests
        • mock
        • ...
      • CGI
      • Fitxers / Files
        • to open files using url, see urlopen
        • io
          • better option than old bultin open (?)
          • read a file:
            • import io

              with io.open(src, 'r', encoding='utf-8') as f_src:
                  for line in f_src:
                      ...
      • URL
        • Parsing URLs with regular expressions
        • 20.5 urllib
          • python2 python3
            urllib
            • urlopen()
            • urlretrieve()
            • ...
            		 urllib2
            • urlopen()
            • install_opener()
            • build_opener()
            • ...
            		 urllib.request
            • urlopen()
            • ...


            		urllib.error
            • exception URLError
            • exception HTTPError
            • exception ContentTooShortError
            urlparse
            • urlparse()
            • parse_qs()
            • parse_qsl()
            • urlunparse()
            • urlsplit()
            • urlunsplit()
            • urljoin()
            • urldefrag()
            urllib
            • quote()
            • quote_plus()
            • unquote()
            • unquote_plus()
            • urlencode()
            • pathname2url()
            • url2pathname()
            • getproxies()
            		urllib.parse
            • urlparse()
            • parse_qs()
            • parse_qsl()
            • urlunparse()
            • urlsplit()
            • urlunsplit()
            • urljoin()
            • urldefrag()
            • unwrap()
            • ...
            • quote()
            • quote_plus()
            • quote_from_bytes()
            • unquote()
            • unquote_plus()
            • unquote_to_bytes()
            • urlencode()


            		urllib.robotparser
          • compatible python 2/3 import
            • import six

              if six.PY2:
                  from urlparse import urlparse, urlunparse
                  from urllib import unquote, urlencode
              else:
                  from urllib.parse import urlparse,               urlunparse, unquote, urlencode

              o = urlparse(...)
              ... unquote(...)
              ... urlunparse(...)
              ... urlencode(...)
          • urlparse()
            • o = urlparse('http://www.cwi.nl:80/%7Eguido/Python.html')
              o
            • o = urlparse('s3://bucket/path/to/file')
              o
          • urlopen()
            • import six

              if six.PY2:
                  from urllib2 import urlopen
                     
                  # to avoid error addinfourl instance has no attribute '__exit__',
                  # on Python 2 you cannot use urlopen inside a with
                  filehandle_src = urlopen(url)
                  for line in filehandle_src:
                      print(line.decode('utf-8'))
                      ...
                     
              else:
                  from urllib.request import urlopen

                  with urlopen(url) as filehandle_src:
                      for line in filehandle_src:
                          ...
          • Changing hostname in a url
            • import urlparse
              p = urlparse.urlparse('https://www.google.dk:80/barbaz')
              p._replace(netloc=p.netloc.replace(p.hostname, 'www.foo.dk')).geturl()
          • URL Quoting (percent encoding)
            • quote
              • import six
                if six.PY2:
                    from urllib import quote
                else:
                    from urllib.parse import quote
              • Python 2
                • from urllib import quote

                  # input is unicode
                  path_original = u'/path/to/toto_à.html'
                  path_converted = quote(path_original.encode('utf8'))
                  # will give: '/path/to/toto_%C3%A0.html'
                • from urllib import quote
                  # input is not unicode
                  path_original = '/path/to/toto_à.html'
                  path_converted = quote(path_original)
                  # will give: '/path/to/toto_%C3%A0.html'
              • Python 3
                • from urllib.parse import quote

                  # input is unicode
                  path_original = u'/path/to/toto_à.html'
                  path_converted = quote(path_original.encode('utf8'))
                  # will give: '/path/to/toto_%C3%A0.html'
                • from urllib.parse import quote

                  # input is not unicode
                  path_original = '/path/to/toto_à.html'
                  path_converted = quote(path_original)
                  # will give: '/path/to/toto_%C3%A0.html
            • unquote
              • convert: /path/to/toto_%C3%A0.mp4 -> /path/to/toto_à.mp4
                • import six

                  path_original = u'/path/to/toto_%C3%A0.mp4'
                  if six.PY2:
                      path_converted = unquote(path_original.encode()).decode('utf-8')
                  else:
                                        path_converted = unquote(path_original)
        • urljoin
        • download from url to local file:
          • import six

            if six.PY2:
                from urlparse import urlparse
                from urllib2 import urlopen
            else:
                from urllib.parse import urlparse
                from urllib.request import urlopen

            with urlopen(src) as filehandle_src, open(dst, 'wb') as filehandle_dst:
                 shutil.copyfileobj(filehandle_src, filehandle_dst)
        • ...
      • Expressions regulars / Regular expressions
        • 7.2. re — Regular expression operations
        • URL parsing
        • Exemple / Example
          • import re
            import dateutil.parser

            regex = re.compile(r'.*_(.*)')

            # get datetime from the following string: 
            name = 'toto_2016-10-25T133700Z'

            m = regex.search(name)
            if m:
                extracted_datetime = dateutil.parser.parse( m.groups()[0] )
        • Remove elements from a list according to a regular expression (files: *.min.js, *.min.css)
          • import re
            list_with_min_js = ['primer.js', 'segon.min.js', 'tercer.js']
            regex = re.compile(r'.*.min.js|.*.min.css')
            list_without_min_js = [i for i in list_with_min_js if not regex.search(i)]
            # returns: ['primer.js', 'tercer.js']
      • JSON
      • CSV
        • 13.1. csv — CSV File Reading and Writing (2.7) (3.7)
        • csv from file
          • with open(csv_path, newline='') as f:
                reader = csv.reader(f, delimiter=';')
                for row in reader:
                    row_length = len(row)
                    print('{} -- {}'.format(row_length, row[10:20]))
          • with open(csv_path, newline='', encoding='Windows-1252') as f:
                reader = csv.reader(f, delimiter=';')
                for row in reader:
                    row_length = len(row)
                    print('{} -- {}'.format(row_length, row[10:20]))
        • csv from string
        • csv from InMemoryUploadedFile:
        • csv from http response (e.g. from APITestCase)
          • import six

            if six.PY2:
                csv_string = res.content
            else:
                csv_string = str(res.content, 'utf-8')
            csv_lines = csv_string.splitlines()
            reader = csv.reader(csv_lines)
            parsed_csv = list(reader)
            number_csv_lines = len(parsed_csv)
      • mime types
        • 18.7 mimetypes
          • used by aws s3 to determine content-type of aws s3 object when uploading it
          • Dependencies
            • CentOS 8
              • sudo dnf install mailcap
                • will install /etc/mime.types
          • Ús / Usage
            • import mimetypes
              guessed_mime_type = mimetypes.guess_type('myfile')
              mimetypes.knownfiles
                                ['/etc/mime.types', '/etc/httpd/mime.types', '/etc/httpd/conf/mime.types', ...]
        • magicfile
          • Dependències / dependencies
            • Mageia
              • urpmi libpython-devel libmagic-devel
            • CentOS
              • sudo yum install file-devel
          • Instal·lació / Installation
            • pip install magicfile
          • Ús / Usage
            • import magicfile as magic
              mime_type = magic.from_file("testdata/test.pdf", mime=True)
            • import magicfile as magic
              f = magic.Magic(magic_file='/usr/share/misc/magic', mime=True)               mime_type = f.from_file("testdata/test.pdf")
      • XML
      • Loops
        • compact
          • ["{}-{}".format(a,b) for a,b in ...]
      • 5.6. Sequence Types — str, unicode, list, tuple, bytearray, buffer, xrange
      • 5.8 Mapping Types - dict
        • items() vs iteritems()
        • Pretty print (not recursive)
          • import pprint
            pp = pprint.PrettyPrinter(width=1)
            pp.pprint(my_dict)
        • json (recursive)

          		• python object (list or dict)
            		 json string
            		 json file
            python object ->
            		 - obj_json = json.dumps(obj, indent=4) with open("toto.json", "w") as f:
                json.dump(obj, f, indent=4)
            json string ->
            		 obj = json.loads(obj_json) - -
            json file ->
            		 with open("toto.json", "r") as f:
                obj = json.load(f)             		- -

          • import json
            print json.dumps(my_dict, indent=2)
        • yaml
          • import yaml
            print yaml.dump(my_dict, indent=2)
      • Data / Date
        • 8.1 datetime
          • 8.1.7 strftime() and strptime() Bahavior
            • now in UTC and ISO-8601 format
              • import datetime
                now = datetime.datetime.utcnow()
                now.isoformat() # 2017-06-14T09:57:56.145575
                now.strftime('%Y-%m-%dT%H:%M:%S.%fZ') # 2017-06-14T09:57:56.145575Z
            • build a datetime from string
              • import datetime
                mydate = datetime.datetime.strptime('20201201T100102', '%Y%m%dT%H%M%S')
            • convert from naive to aware:
              • import datetime
                import pytz

                mydate = datetime.datetime.strptime('2020-11-01T00:00:00', '%Y-%m-%dT%H:%M:%S')

                # convert datetime from naive to aware (utc):
                utc = pytz.utc
                mydate_utc = utc.localize(mydate)
            • convert from ISO-8601 string to datetime
              • import dateutil.parser
                yourdate = dateutil.parser.parse(datestring)
              • my_time = strptime( my_string, '...')
            • convert seconds to HH:MM:SS
              • without second fractions
                • import time

                  time_in_hh_mm_ss = time.strftime('%H:%M:%S', time.gmtime(time_in_seconds_int))
              • with second fractions
                • # similar to timedelta.__str__
                  def _to_hh_mm_ss_ms(seconds):
                      mm, ss = divmod(seconds, 60)
                      hh, mm = divmod(mm, 60)
                      fraction_seconds = seconds - int(seconds)
                      s = "{:d}:{:02d}:{:02d}.{:03d}".format(hh, mm, ss, fraction_seconds)
                      return s
                • import datetime

                  # no fractional part if it is 0
                  time_in_hh_mm_ss = str(datetime.timedelta(seconds=duration_seconds_float))
                • import datetime
                  import pytz

                  time_in_hh_mm_ss_ff = datetime.datetime.fromtimestamp(time_in_seconds_float, pytz.UTC).strftime('%H:%M:%S.%f')
            • convert HH:MM:SS to seconds
            • ...
        • 15.3 time - Time access and conversions
        • minimum and maximum datetime (infinite)
          • unaware_maximum = datetime.datetime.max
          • import pytz
            aware_maximum = datetime.datetime.max.replace(tzinfo=pytz.UTC)
        • other libraries
        • add datetime and time
          • event.start + datetime.timedelta(hours=event.duration.hour,minutes=event.duration.minute,seconds=event.duration.second)
        • difference between two times
        • Date in Django
        • Periods
        • Fusos horaris / Timezones
          • import pytz
            timezone = 'Europe/Andorra'
            my_date.astimezone(pytz.timezone(timezone)).strftime('%H:%M:%S %Z')
      • Adreces de xarxa / Network addresses
        • adreça IP pròpia / own IP address
        • 21.28 ipaddress (>3.3)
          • Backport for Python 2: py2-ipaddress
          • An introduction to the ipaddress module
          • models


            		• convenience factory function
              		 functions
              IPv4Address IPv6Address ip_address('192.168.1.100')
              IPv4Network IPv6Network ip_network('192.168.1.0/24')
              • subnets()
              • subnets(prefixlen_diff=2)
              • subnets(new_prefix=26)
              • supernet()
              • supernet(prefixlen_diff=2)
              • supernet(new_prefix=26)
              IPv4Interface IPv6Interface
              		ip_interface('192.168.1.100/24')
      • Random
        • 9.6 random
        • Contrasenya aleatòria / Random password
          • Generate password in python
          • import random
            import string

            chars = string.letters + string.digits
            length = 20
            generated_password = ''.join(map(lambda x: random.choice(chars), range(length)))
  • Unicode
    • Python3 strings and bytes
    • Solving Unicode Problems in Python 2.7
      • UnicodeDecodeError: ‘ascii’ codec can’t decode byte 0xd1 in position 1: ordinal not in range(128) (Why is this so hard??)
    • Django i18n
      • Use format instead of '%'
        • do not forget trailing 'u'
          • u'{} {} '.format(...)
      • Use 'u' before literals
      •     string = u'{:02d}:{:02d}:{:02d}'.format(hours, minutes, seconds)
            if days:
                if days==1:
                    string = u'{} {} '.format(days,_("day")) + string
                    #string = u"%d %s %s" % ( days, _("day"), string)
                else:
                    string = u'{} {} '.format(days,_("days")) + string
                    #string = u"%d %s %s" % ( days, _("days"), string)
    • File writing
      • import codecs
        filename = os.path.join(...)
        f = codecs.open(filename,mode='w',encoding='utf-8')
        f.write(content)
        f.close        ()
      • import codecs
        filename = os.path.join(...)
        f = codecs.open(filename,mode='w',encoding='utf-8')
        fitxer = File(f)
        fitxer.write(content)
        fitxer.close
  • Eines / Tools
    • virtualenv
    • pip (*) (package manager) (also installed when virtualenv is installed)
      • related tools
        • pipdeptree
        • pipupgrade
      • Installation of pip itself
        • From distribution
          • CentOS
            • sudo yum install python-pip
        • From source
          • download it:
          • install it:
            • # python setup.py install
      • Installation of packages
        • pip install package_name
      • Installation of a precise version of a package
        • pip install djangorestframework==0.4.0
      • alpha version
        • pip install -pre package_name
      • upgrade
        • pip install -U package_name
      • Problems
        • error fatal: Python.h: El fitxer o directori no existeix
          • Solució / Solution
            • Python 2.7
              • Mageia
                • urpmi lib64python-devel
            • Python 3
              • Mageia
                • urpmi lib64python3-devel
        • Download error on https://pypi.python.org/simple/: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed (_ssl.c:765) -- Some packages may not be found!
          • pip install fails with “connection error: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed (_ssl.c:598)”
          • Solution
            • openssl s_client -connect pypi.python.org:443
            • curl -sO http://cacerts.digicert.com/DigiCertHighAssuranceEVRootCA.crt
            • sudo cp DigiCertHighAssuranceEVRootCA.crt /etc/pki/ca-trust/source/anchors/
            • sudo update-ca-trust
          • Alternative solution?
            • sudo yum install ca-certificates
        • pip install --upgrade -r pip_requirements.txt
          • Could not find .egg-info directory in install record for setuptools from https://pypi.python.org/packages/25/4e/1b16cfe90856235a13872a6641278c862e4143887d11a12ac4905081197f/setuptools-28.8.0.tar.gz#md5=43d6eb25f60e8a2682a8f826ce9e3f42 in /home/.../env/lib/python2.7/site-packages
          • see also problems with Google API and httplib2
        • error: Installed distribution setuptools 0.9.8 conflicts with requirement setuptools>=17.1
          • Solució / Solution
            • pip install --upgrade setuptools
        • Error: pg_config executable not found.
          • Solució / Solution
            • Install postgresql devel
              • Mageia
                • urpmi postgresql9.4-devel
      • list of installed packages (it gives the version number):
        • pip freeze
    • easy_install
      • urpmi python-setuptools
      • sudo apt-get install python-setuptools
    • Invoke
    • Fabric
      • Instal·lació / Installation
        • pip install fabric
      • Fabric 2.x
        • See also: Invoke
        • Upgrading from 1.x
          • 1.x
            		 2.x
            sudo("yum install -y htop") c.run("sudo yum install -y htop")
            sudo("echo /usr/local/lib >/etc/ld.so.conf.d/local.conf") c.run("sudo sh -c 'echo \"/usr/local/lib\" >/etc/ld.so.conf.d/local.conf'")
            # sudo sh -c 'echo "l'\''arbre" >>/tmp/toto.txt'
            c.run("sudo sh -c 'echo \"l'\\''arbre\" >>/tmp/toto.txt'")
            put(local_file, remote_dir, use_sudo=True)
            		 from os.path import basename

            def sudo_put(c, local_path, remote_dirname):
                """
                Upload a local file to a remote dir, with sudo privileges
                """
                filename = basename(local_path)
                remote_tmp_path = filename
                remote_path = '{}/{}'.format(remote_dirname, filename)

                print 'sudo_put: {} -> {}'.format(local_path, remote_path)
               
                c.put(local_path, remote=remote_tmp_path)
                c.run("sudo sh -c 'mv {} {}'".format(remote_tmp_path, remote_path))

            ...
            sudo_put(c, local_file, remote_dir)
            from fabric.contrib.files import append
            text = """
            FCGI_EXTRA_OPTIONS="-M 0770"
            """
            append('/etc/sysconfig/spawn-fcgi', text, use_sudo=True)
            		 text = """
            FCGI_EXTRA_OPTIONS=\\\"-M 0770\\\"
            """
            c.run("sudo sh -c 'echo \"{0}\" >>/etc/sysconfig/spawn-fcgi'".format(text))

            # not working yet, with sudo=True:
            from patchwork import files
            files.append(c, '/etc/sysconfig/spawn-fcgi', text, sudo=True)
            from fabric.contrib.files import sed

            sed('{}/pg_hba.conf'.format(pgsql_data_dir),
                'host    all             all             127.0.0.1/32            ident',
                'host    all             all             127.0.0.1/32            md5',
                use_sudo=True)            		 c.run("sudo sed -i.bak 's#host    all             all             127.0.0.1/32            ident#host    all             all             127.0.0.1/32            md5#g' {}/pg_hba.conf".format(pgsql_data_dir))
        • Documentation (2.1)
      • Fabric 1.x documentation
      • Utilització / Usage
        • <task name>:<arg>,<kwarg>=<value>,...
      • debug from Eclipse
        • fabfile.py
          • ...
            from fabric.main import main

            if __name__ == '__main__':
                import sys
                sys.argv = ['fab', '-f', __file__, 'my_task']
                main()
      • Exemples / Examples
        • put('toto.sh', '/usr/local/bin/', mode=int('755', 8), use_sudo=True)
      • Context managers
        • with ...
      • Problemes / Problems
        • paramiko.ssh_exception.SSHException: encountered RSA key, expected OPENSSH key
          • check that remote file ~myuser/.ssh/authorized_keys contains the public key that you are using
          • check that you are specifying a remote user
      • Error management
        • ignore
          • with settings(warn_only=True):
        • capture failure
          • result = local('grunt')
            if result.failed:
                print "Grunt is not installed."
                abort('Grunt not found')
        • Python try
          • try:
                sudo('...')
            except Exception as e:
                ...
                abort('...')
      • env definitions
        • fabfile.py
          • from fabric.api import *

            # remote user and group
            env.remote_user = 'myuser'
            env.remote_group = 'mygroup'

            # project
            env.project_name = 'my_project'
            env.project_dir = '/home/%(remote_user)s/%(project_name)s' % env
      • run locally
        • running fabric script locally
        • Optionally avoid using ssh if going to localhost #98
        • fabfile.py
          • # by default, actions are performed remotely.
            # to perform them localy, e.g.: "fab localhost create_virtualenv"
            env.run_as = 'remote'
            env.run = run
            env.sudo = sudo
            env.hosts = ['...',]
            env.key_filename = '~/.ssh/keys/key_for_remote.pem'

            def localhost():
                """
                Set environment for local execution
                """
                env.run_as = 'local'
                env.run = local
                env.sudo = local
                env.hosts = []
          • with lcd()
                ...
      • capture
        • run remotelly and get the value
          • result = sudo(...)
          • result = run(...)
        • run locally and get the value
          • result = local(..., capture=True)
      • crontab
      • files
        • put
        • rsync_project
          • How do I copy a directory to a remote machine using Fabric?
          • Example
            • from fabric.contrib.project import rsync_project

              rsync_project(local_dir='.', remote_dir='/var/www', exclude=('.git','tmp',) )
          • Ignore files indicated by .gitignore
          • Ignore untracked files and files indicated by .gitignore
            • untracked_files_zero = local('git -C .. ls-files -z -o --exclude-standard --directory', capture=True)
              untracked_files = untracked_files_zero.split('\0')
              print "untracked_files: {}".format(untracked_files)
              excluded_files = untracked_files + ['.git','tmp']
              rsync_project(local_dir='.', remote_dir=env.project_dir, exclude=excluded_files, extra_opts="--filter=':- .gitignore'" )
        • append
          • append a line
            • from fabric.contrib.files import append
              ...
                  append('/etc/sysconfig/toto','this line has been appended to the end')
          • append several lines:
            • from fabric.contrib.files import append
              ...
                  text = """
              first_line = "value1"
              second_line = "value2"
              """
                  append('/etc/sysconfig/toto', text, use_sudo=True)
        • sed
          • single quotes in sed
          • replace a line that starts with "host    all             all             127.0.0.1" by "host    all             all             127.0.0.1/32            md5"
            • from fabric.contrib.files import sed
              ...
                  # /var/lib/pgsql/data/pg_hba.conf
                  # host    all             all             127.0.0.1/32            md5
                  sed('/var/lib/pgsql/data/pg_hba.conf',
                      'host    all             all             127.0.0.1/32            ident',
                      'host    all             all             127.0.0.1/32            md5',
                      use_sudo=True )
            •     env.sudo('sudo sed -e "/^host    all             all             127.0.0.1/ c\host    all             all             127.0.0.1/32            md5" -i /var/lib/pgsql/data/pg_hba.conf')
      • Certificat de servidor / Server certificate
        • certificate for https connections (curl will not need to specify --cacert)
          • # add myserver self-signed certificate to the list of trust ca certificates
            put('myserver.example.org.crt', '/etc/pki/ca-trust/source/anchors/', use_sudo=True)
            env.sudo('update-ca-trust')
      • dependencies
        • fabfile.py
          • # list of dependencies to install
            env.install_cmd = 'yum install -y'
            env.dependencies = ['gcc', 'git', 'python-virtualenv', 'mariadb',]

            def install_dependencies():
                """
                Install the system dependencies for the project
                """
                env.sudo(env.install_cmd + " " + "epel-release")
                env.sudo(env.install_cmd + " " + " ".join(env.dependencies))
      • virtualenv
        • Getting virtualenv(wrapper) and Fabric to play nice
        • Activate a virtualenv via fabric as deploy user
        • fabfile.py
          • from contextlib import contextmanager as _contextmanager

            # remote user and group
            env.remote_user = 'my_user'
            env.remote_group = 'my_group'

            # virualenv directory
            env.virtualenv_directory = '/opt/p27'
            env.virtualenv_activate = 'source %(virtualenv_directory)s/bin/activate' % env

            def create_virtualenv():
                """
                Create the virtualenv
                """
                env.sudo('mkdir -p %(virtualenv_directory)s' % env)
                env.sudo('chown %(remote_user)s.%(remote_group)s %(virtualenv_directory)s' % (env) )
                env.run('virtualenv %(virtualenv_directory)s' % env)

            @_contextmanager
            def virtualenv():
                """
                Activate the virtualenv
                """
                with cd(env.virtualenv_directory):
                    with prefix(env.virtualenv_activate):
                        yield

            def install_pip_dependencies():
                """
                Install the pip dependencies
                """
                with virtualenv():
                    if env.run_as == 'remote':
                        put('pip_requirements.txt', 'pip_requirements.txt')
                    env.run('pip install -r pip_requirements.txt')
      • Django
        • fabfile.py
          • def django_setup():
                """
                Django: migrate, createsuperuser, collectstatic
                """
                with virtualenv():
                    with cd(env.project_dir):
                        env.run('python manage.py migrate')
                        env.run('python manage.py createsuperuser')
                        env.run('python manage.py collectstatic')
                        env.run('python manage.py loaddata auth_initial')

            def django_update():
                """
                Django: migrate, collectstatic
                """
                with virtualenv():
                    with cd(env.project_dir):
                        env.run('python manage.py migrate')
                        env.run('python manage.py collectstatic')
      • Nginx
        • fabfile.py
          • def nginx_setup():
                """
                Configure and start nginx
                """
                env.sudo('chmod 755 /home/centos')
                env.sudo('mkdir -p /etc/uwsgi/vassals/')
                if env.run_as == 'remote':
                    # nginx
                    put('nginx-uwsgi/%(project_name)s_nginx.conf' % env, '/etc/nginx/conf.d/', use_sudo=True)
                    # remove default site from nginx.conf
                    put('nginx-uwsgi/nginx.conf', '/etc/nginx/', use_sudo=True)
                    put('nginx-uwsgi/nginx.pp', '/etc/nginx/', use_sudo=True)
                   
                    # uwsgi       
                    put('nginx-uwsgi/uwsgi_params', '/etc/uwsgi/', use_sudo=True)
                    put('nginx-uwsgi/emperor.ini', '/etc/uwsgi/', use_sudo=True)
                    put('nginx-uwsgi/%(project_name)s_uwsgi.ini' % env, '/etc/uwsgi/vassals/', use_sudo=True)
                    put('nginx-uwsgi/emperor.uwsgi.service', '/etc/systemd/system/', use_sudo=True)
                    # socket in /run/ (http://uwsgi-docs.readthedocs.org/en/latest/Systemd.html#putting-sockets-in-run)
                    #put('nginx-uwsgi/emperor.uwsgi.socket', '/etc/systemd/system/', use_sudo=True)
                    #put('nginx-uwsgi/emperor.uwsgi.conf', '/etc/tmpfiles.d/', use_sudo=True)
               
                # custom selinux policy module (http://axilleas.me/en/blog/2013/selinux-policy-for-nginx-and-gitlab-unix-socket-in-fedora-19/)
                env.sudo('semodule -i /etc/nginx/nginx.pp')
                # activate selinux
                env.sudo('setenforce 1')

                # enable and start nginx
                env.sudo('systemctl enable nginx.service')
                env.sudo('systemctl restart nginx.service')
              
                # enable and start uwsgi
                env.sudo('systemctl enable emperor.uwsgi.service')
                env.sudo('systemctl restart emperor.uwsgi.service')
              
                # configure firewall
                #env.sudo('firewall-cmd --permanent --zone=public --add-service=http')
                #env.sudo('firewall-cmd --permanent --zone=public --add-service=https')
                #env.sudo('firewall-cmd --reload')


            def nginx_restart():
                """
                Restart nginx and wsgi
                """

                # restart nginx
                env.sudo('systemctl restart nginx.service')
              
                # restart uwsgi
                env.sudo('systemctl restart emperor.uwsgi.service')
      • Database
        • fabfile.py
          • # database
            env.mysql_host = 'localhost'
            env.mysql_database = 'mydatabase_db'
            env.mysql_user = 'my_user'
            env.mysql_password = 'my_password'
            env.mysql_master_user = 'root'

            def database_setup():
                """
                Setup database service
                """
                env.sudo("systemctl enable mariadb.service")
                env.sudo("systemctl start mariadb.service")
                env.sudo("mysql_secure_installation")


            def database_create():
                """
                Create the sql database
                """
                env.run('echo "CREATE DATABASE IF NOT EXISTS %(mysql_database)s; \
            GRANT ALL ON %(mysql_database)s.* TO \'%(mysql_user)s\'@\'%%\' IDENTIFIED BY \'%(mysql_password)s\'; \
            FLUSH PRIVILEGES;" | \
            mysql -h %(mysql_host)s -u %(mysql_master_user)s -p' % (env) )


            def database_delete():
                """
                Delete the sql database
                """
                env.run('echo "DROP DATABASE %(mysql_database)s;" | \
            mysql -h %(mysql_host)s -u %(mysql_master_user)s -p' % (env) )
      • Git
        • fabfile.py
          • def ssh_config():
                """
                Add fabuser_bitbucket_support to ~/.ssh/config
                """
                text = """
            Host fabuser-bitbucket
                 HostName bitbucket.org
                 IdentityFile ~/.ssh/fabuser_bitbucket
                """
                append('%s/config' % env.ssh_dir, text )
                env.run('chmod 600 %s/config' % env.ssh_dir)

            def git_clone():
                """
                Clone from git
                """
                #git_user = 'francesc_pinyol_margalef'
                with settings(warn_only=True):
                    with settings(warn_only=True):
                        if env.run("test -d %s" % env.project_dir).failed:
                            env.run("git clone git@fabuser-bitbucket:%(bitbucket_account)s/%(project_name)s.git %(project_dir)s" % (env) )


            def git_pull():
                """
                Pull from git
                """
                with cd(env.project_dir):
                    env.run("git pull")
    • Empaquetament / Packaging
  • Multiple platforms
  • Python path:
    • see also bash path
    • /usr/local/lib/python2.7/dist-packages/...
    • paths
      • import os

        my_path = './dir1/dir2/toto.mp4'      # ./dir1/dir2/toto.mp4
        my_dirname = os.path.dirname(my_path)    # ./dir1/dir2
        #my_rel_dirname=${my_dirname#*\./} # dir1/dir2
        my_basename = os.path.basename(my_path)  # toto.mp4
        my_name = os.path.splitext(my_basename)[0]    # toto
        my_extension = os.path.splitext(my_path)[1]   # .mp4
        my_rel_path = os.path.relpath(my_path)       # dir1/dir2/toto.mp4
        my_abs_path = os.path.abspath(my_path)  # /path/to/dir1/dir2/toto.mp4
    • print path:
      • python
        • import sys
          sys.path
    • set path:
      • python
        • import sys
          sys.path.append("/my/path")
      • Django:
        • /etc/apache2/mods-available/wsgi.conf
          • WSGIPythonPath ...
    • recursively create directory if it does not exist:
      • import os

        # create the directory if it does not exist
        father_dir = os.path.dirname(filename)
        if not os.path.exists(father_dir):
            os.makedirs(father_dir)
            print("creating directory: {}".format(father_dir))
    • get home dir:
      • from os.path import expanduser
        home = expanduser("~")
    • get absolute dir inside home:
      • from os.path import expanduser
        my_dir = expanduser("~/my_dir")
  • Python for MS Windows:
  • HTML parsing
  • Logging
    • Django logging
    • Logging Cookbook
    • Exemple / Example:
      • # logger
        import logging

        logger = logging.getLogger('my_program')
        logger.setLevel(logging.DEBUG)
        # create console handler with a higher log level
        ch = logging.StreamHandler()
        ch.setLevel(logging.DEBUG)
        # create formatter and add it to the handlers
        formatter = logging.Formatter('%(asctime)s %(levelname)-8s %(name)-12s %(message)s', '%Y-%m-%dT%H:%M:%SZ')
        ch.setFormatter(formatter)
        # add the handlers to the logger
        logger.addHandler(ch)

        logger.debug("my message")
  • Nginx

Pylint
  • Pylint documentation
    • Instal·lació / Installation
      • pip install pylint
        • Installing collected packages: backports.functools-lru-cache, isort, mccabe, singledispatch, wrapt, lazy-object-proxy, astroid, pylint
          Successfully installed astroid-1.6.6 backports.functools-lru-cache-1.6.4 isort-4.3.21 lazy-object-proxy-1.6.0 mccabe-0.6.1 pylint-1.9.5 singledispatch-3.6.2 wrapt-1.12.1
      • to use with Django
        • to avoid errors: "E: xx,yy: Class '...' has no 'objects' member (no-member)"
        • install pylint-django:
          • pip install pylint-django
          • Python 2.7
            • $ pip install "pylint-django<2"
              Successfully built pylint-django pylint-plugin-utils
              Installing collected packages: pylint-plugin-utils, pylint-django
              Successfully installed pylint-django-0.11.1 pylint-plugin-utils-0.6
    • IDE integration
    • Run
      • pylint mymodule
      • to use with Django code
        • DJANGO_SETTINGS_MODULE=your.app.settings pylint --load-plugins pylint_django [..other options..] <path_to_your_sources>
      • only some checkers (Error, Warning, ...)
        • pylint --disable=all --enable=E,W mymodule
        • pylint --disable=all --enable=unused-import mymodule
      • help about a message
        • pylint --help-msg=no-member
      • logging
      • ...

Tox
  • Tox
  • Instal·lació / Installation
    • System-wide
      • Mageia
        • sudo urpmi python2-tox python3-tox
  • Totes les versions de Python que especifiqueu, han d'estar instal·lades
  • Tutorials
  • Django
    • Testing a third party Django application with pytest and tox
    • Exemples / Examples
      • [tox]
        envlist =
            django111-py{27,36,37}
        skipsdist = true

        [testenv]
        commands =
             {envpython} -m manage test --settings my_project.tests_settings --verbosity 3 my_app
        deps =
             django111-py27: -r deployment/pip_requirements.py27.txt
             django111-py36: -r deployment/pip_requirements.py36.txt
             django111-py37: -r deployment/pip_requirements.py37.txt
  • Ús / Usage
    • list available environments
      • tox -l
    • run all environments
      • tox
    • run a specific environment
      • tox -e django111-py27

Biblioteques / Libraries
  • Useful modules
    • Dades / Data
    • Bases de dades / Databases
    • Imatges / Images
    • Criptografia / Cryptography
    • Autenticació / Authentication
    • Xarxa / Network
    • Processos / Processes
      • 17.1 subprocess (2) (3)
        • subprocess.run (>=3.5)
        • Older high-level API
          • old
            		 new
            subprocess.call(...) subprocess.run(...)
            subprocess.check_call(...) subprocess.run(..., check=True)
            subprocess.check_output(...) subprocess.run(..., check=True, stdout=subprocess.PIPE).stdout
      • Exemples / Examples
        • compatible Python 2 / 3:
          • import shlex
            if six.PY2:
                # python 2
                # https://stackoverflow.com/questions/14218992/shlex-split-still-not-supporting-unicode#answer-14219159
                args = map(lambda s: s.decode('utf-8'), shlex.split(complete_command.encode('utf-8')))
                result_stdout = subprocess.check_output(args, stderr=subprocess.STDOUT)
                result_stderr = ''
            else:
                # python 3
                args = shlex.split(complete_command)
                completed_process = subprocess.run(args, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                result_stdout = completed_process.stdout
                result_stderr = completed_process.stderr
        • standard
          • import shlex, subprocess

            command = '...'
            parameters = '...'
            command_line = "{0} {1}".format(command, parameters)             args = shlex.split(command_line)

            try:
                completed_process = subprocess.run(args, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                print(completed_process.stdout.decode('utf-8'))
                print(completed_process.stderr.decode('utf-8')) except Exception as e:
                print("ERROR: {0}".format(e))
                if e.stdout:
                    print("ERROR stdout: {0}".format(e.stdout.decode('utf-8')))
                if e.stderr:
                    print("ERROR stderr: {0}".format(e.stderr.decode('utf-8')))
        • stdout, stderr to a file:
          • import shlex, subprocess

            command = '/usr/bin/ffmpeg'
            parameters = '-y -i sintel.mp4 -vf "scale=1280:-1" -c:a copy -c:v h264 -f flv /tmp/toto.flv'
            command_line = "{0} {1}".format(command, parameters)             args = shlex.split(command_line)
            with open("toto.log", "w") as f:
                try:
                    completed_process = subprocess.run(args, check=True, stdout=f, stderr=f)
                except Exception as e:
                    print("ERROR: {0}".format(e))
      • Problemes / Problems
    • SSH
      • Paramiko
        • Documentation
        • Example
          • import paramiko

            client = paramiko.SSHClient()
            #client.load_system_host_keys()
            k = paramiko.RSAKey.from_private_key_file('/home/my_user/.ssh/keys/remoteserver.pem')
            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            client.connect('remoteserver.example.org', username='myremoteuser', pkey=k)
            ssh_stdin, ssh_stdout, ssh_stderr = client.exec_command('ls -ltr')

            for line in ssh_stdout:
                print line
    • SVG
      • Creating Simple SVG from Python
      • pySVG
      • svgwrite
        • Documentation
        • mozman / svgwrite (github)
        • Exemples / Examples
          • create an empty document 
            • #!/usr/bin/env python3
# -*- coding: utf-8 -*-

import svgwrite
from svgwrite import mm
from svgwrite.container import Group

# create svg document
svg_width = 100 # mm
svg_height = 200 # mm
path = "toto.svg"
dwg = svgwrite.Drawing(path, profile='full', size=(svg_width*mm, svg_height*mm))

dwg.viewbox(0, 0, svg_width, svg_height)

# save svg to file
dwg.save(pretty=True)
          • ...
        • utf-8
          • # save svg to file
            from xml.etree.ElementTree import tostring
            dwg.tostring = lambda self=dwg: tostring(self.get_xml()).decode("utf-8")
            dwg.save(pretty=True)
      • convert svg to image
        • Convert SVG to PNG in Python
        • CairoSVG (Python3)
          • install Python3 and virtualenv
          • pip3 install cairosvg
          • as command:
            • cairosvg image.svg [--output_width 200] -o image.png
          • as library:
            • import cairosvg
              cairosvg.svg2pdf(url='image.svg',               output_width=200, write_to='image.pdf')
        • pyrsvg (librsvg)
          • pyrsvg How to use librsvg from Python (Cairo)
          • Installation
            • Mageia
              • urpmi gnome-python-desktop
                • /usr/lib64/python2.7/site-packages/gtk-2.0/rsvg.so
                • /usr/share/doc/gnome-python-desktop/rsvg/rsvg-cairo.py
            • Ubuntu
          • Ús / Usage
            • virtualenv --system-site-packages env
            • souce env/bin/activate
            • python toto.py
            • toto.py
              • import cairo
                import rsvg
                ...
    • FTP
      • libftp (2.7)
        • upload local or remote file to ftp:
          • # python 3
            import os
            import ftplib
            import urllib.request
            from urllib.parse import urlparse

            host = '...'
            username = '...'
            password = '...'

            #origin = '/tmp/toto.txt'
            origin = 'https://...'
            local_basename = os.path.basename(origin)
            remote_dirname = 'toto_dir_2'
            remote_path = os.path.join(remote_dirname, local_basename)

            print("ftp upload: {0} -> {1}".format(origin, remote_path))

            o = urlparse(origin)
            print("sheme: {}".format(o.scheme))

            if o.scheme:
                filehandle = urllib.request.urlopen(origin)
            else:
                filehandle = open(origin, 'rb')


            with ftplib.FTP(host, username, password) as ftp:
                # create dir
                try:
                    print("creating remote dir: {}".format(remote_dirname))
                    ftp.mkd(remote_dirname)
                except Exception as e:
                    print("WARNING when creating dir: {} - {}".format(e, type(e)))

                # upload file
                try:
                    print("uploading: {0} -> {1}".format(origin, remote_path))
                    ftp.storbinary("STOR {0}".format(remote_path), filehandle)
                except Exception as e:
                    print("ERROR when uploading file: {}".format(e))

            print("done")
        • upload remote http file directly to ftp (Python - Transfer a file from HTTP(S) URL to FTP/Dropbox without disk writing (chunked upload)):
          • ...
            import urllib
            ...
            filehandle = urllib.request.urlopen(origin_url)
            ...
    • HTTP / HTTPS
          • post
            • import requests

              url = ...
              payload = {'toto_key':'toto_value',}
              r = requests.post(url, payload)
          • jwt + get
            • import requests
              from pprint import pprint

              backend_url = 'http://127.0.0.1:8000'

              # jwt token
              # file toto_admin.txt contains the password
              password = open('toto_admin.txt', 'r').read().strip()
              payload_credentials = {
                                     'username': 'admin',
                                     'password': password
                                     }
              r = requests.post(backend_url+'/api-token-auth/', data=payload_credentials)
              token = r.json()['token']

              r = requests.get('%s/v1/api/totos/' % (backend_url), headers={'Authorization':'JWT %s'%token})
              pprint( r.json() )
          • upload (e.g. with Django Restframework)
            • POST a Multipart-Encoded File
            • Exemple / Example:
              • models.py
                • class MyModel(models.Model):
                      field_1 = models.FileField(...)
                      field_2 = ...
                      field_3 = ...
              • upload_test.py
                • import requests

                  url = 'http...'
                  payload = {'field_2': '...', 'field_3': '...'}
                  headers = {}
                  headers['...'] = '...'
                  with                   open('/tmp/toto.png', 'rb') as field_1_file:   
                      files = {'field_1': field_1_file}
                      # to specify a name
                      files = {'field_1': ('name of the uploaded file', field_1_file)}
                        r = requests.post(url, payload, headers=headers, files=files)
                • # upload from a url
                  source_url = ...

                  url = ...
                  payload = {'field_2': '...', 'field_3': '...'}
                  headers = {}
                  files = {'field_1': ('name of the uploaded file', urlopen(source_url))}
                  r = requests.post(url, payload, headers=headers, files=files)
          • HTTPS
        • Problemes / Problems
          • empty request.DATA when receiving a put/patch:
            • Solució / Solution
              • check that there is a trailing "/". Otherwise, requests receives a 30x and data is lost on the second url
          • claudàtors en el payload / square brackets in payload
            • Solució / Solution: explicit conversion to json, with json header
              • import requests
                import json

                payload = {
                           'toto':['first element','second element'],
                           }
                r = requests.post(address, headers={'Content-Type':'application/json'}, data=json.dumps(payload))
              • import requests
                import json

                # get token as usual
                ...
                token = ...

                payload = {
                           'toto':['first element','second element'],
                           }
                r = requests.post(address, headers={'Authorization':'JWT %s'%token, 'Content-Type':'application/json'}, data=json.dumps(payload))
      • pycurl (curl)
        • Exemple HHTPS / HTTPS Example
          • import pycurl
            curl = pycurl.Curl()
            curl.setopt(pycurl.CAINFO, "ca.crt")
            curl.setopt(pycurl.SSL_VERIFYPEER, 1)
            curl.setopt(pycurl.SSL_VERIFYHOST, 2)
            curl.setopt(pycurl.URL, "https://server_name/")
            curl.perform()
  • PyGObject (based on GObject)

    		• C
      		 Python
      • import gi
        gi.require_version("Gtk", "3.0")
        from gi.repository import Gtk
      		error when packages are not installed

      		-
      		 PyGObject
      • virtualenv: pip install pygobject
        • dependències / dependencies:
          • urpmi lib64girepository-devel gobject-introspection libcairo-devel [?python3-cairo-devel]
          • yum install gcc gobject-introspection-devel cairo-gobject-devel freetype-devel
          • apt-get ...
      • urpmi [python-gobject] python3-gobject3
      • yum install python34-gobject
      • apt-get ...
      • ModuleNotFoundError: No module named 'gi'
      • Package cairo was not found in the pkg-config search path.
      GObject libraries:
      • /usr/lib[64]/girepository-1.0/
      • GI_TYPELIB_PATH
      		GLib
      • GLib-2.0.typelib


      Gtk
      • Gtk-2.0.typelib
      • Gtk-3.0.typelib


      GStreamer
      • Gst-1.0.typelib
      • Gst...1.0.typelib
      • GES-1.0.typelib
      • ValueError: Namespace Gst not available
      • ValueError: Namespace GstWebRTC not available
      ...

    • Instal·lació / Installation
      • Sistema / System
        • CentOS
          • sudo yum install python36-gobject
        • Mageia
          • urpmi ...
      • Virtualenv + pip
        • Dependències / Dependencies
          • Install virtualenv
          • CentOS
            • sudo yum install gobject-introspection-devel cairo-gobject-devel freetype-devel
          • Mageia
            • urpmi lib64girepository-devel gobject-introspection
        • virtualenv-3.5 env (CentOS: virtualenv-3 --python python36 env)
        • source env/bin/activate
        • [pip install --upgrade pip]
        • pip install pygobject
    • Libraries are retrieved from:
      • standard location:
        • /usr/lib64/girepository-1.0/*.typelib
      • non-standard location (e.g. GStreamer compiled from source and installed into /usr/local)
        • export GI_TYPELIB_PATH=/usr/local/lib/girepository-1.0
    • PyGObject API Reference
    • Exemple / Example
      • import gi
        gi.require_version("Gtk", "3.0")
        from gi.repository import Gtk

        window = Gtk.Window(title="Hello World")
        window.show()
        window.connect("destroy", Gtk.main_quit)
        Gtk.main()
    • Problemes
      • import gi
        ModuleNotFoundError: No module named 'gi'
        • Solució / Solution
          • Install Python 3.6 bindings for GObject Introspection
            • sudo yum install python36-gobject
            • ...
      • ValueError: Namespace Gst not available
        • Solució / Solution
          • check that Gst is installed in standard location:
            • ls -l /usr/lib64/girepository-1.0/Gst*.typelib
          • if not, check that Gst is installed in non-standard location:
            • ls -l /usr/local/lib/girepository-1.0/Gst*.typelib
            • add non-standard location to search path:
              • export GI_TYPELIB_PATH=/usr/local/lib/girepository-1.0
  • Parsing
  • Fulls de càlcul / Spreadsheet

    		• CSV OpenPyXL
      read each row as an OrderedDict
      A_label B_label C_label
      A2_value B2_value C2_value
      A3_value B3_value C3_value
      		import csv

      with open(spreadsheet_path, newline='', encoding=file_encoding) as f:
          reader = csv.DictReader(f, delimiter=';')
          print(reader.fieldnames)
          for row in reader:
              print(row['C_label'])
              print(row['B_label'])
              print(row['A_label'])      		 import openpyxl

      wb = openpyxl.load_workbook(spreadsheet_path)
      for sheet_name in wb.sheetnames:
          ws = wb[sheet_name]
          # header
          fieldnames = []
          for column in ws.iter_cols(1, ws.max_column):
              # header value corresponds to the cell on the top ([0])
              fieldnames.append(column[0].value)
              print(fieldnames)
              for row_cells in ws.iter_rows(min_row=2):
                  row = OrderedDict()
                  column_index = 0
                  for cell in row_cells:
                      row[fieldnames[column_index]] = cell.value
                      column_index += 1
                        print(row['C_label'])
                  print(row['B_label'])
                  print(row['A_label'])






    • CSV
    • OpenPyXL
      • print all rows and columns from a binary file retrieved using requests (or unit test)
        • from __future__ import print_function

                  file_like_object = BytesIO(res.content)
                  wb = openpyxl.load_workbook(file_like_object)
                  for sheet_name in wb.sheetnames:
                      print(u"- sheet: {}".format(sheet_name))
                      ws = wb.get_sheet_by_name(sheet_name)
                      for row in ws.iter_rows():
                          for cell in row:
                              print(cell.value, end=',')
                          print()
        • ...
    • xlwt (used by Django XMLRenderer)
    • ...
  • Microsoft SharePoint
  • ...

Celery
  • Celery with Django
  • Arquitectura / Architecture
    • Setting up a queue service: Django, RabbitMQ, Celery on AWS
    • Task Routing in Celery - How to route a Celery task to a dedicated queue
    • Dynamic Task Routing in Celery
    • Specify Worker in Celery
    • Elements
      • broker:
        • controls queues
        • RabbitMQ
          • sudo systemctl start rabbitmq-server.service
          • rabbitmqctl list_users
          • rabbitmqctl list_vhosts
      • worker:
        • executes an async function
        • connects to broker specified as Celery.broker in mydir/celery.py (-A mydir)
          • Problems
            • consumer: Cannot connect to amqp://...: [Errno 13] Permission denied.
              • Solution: check SELinux on instance that is initiating the connection
        • ask broker for the list of active nodes and tasks:
          • .../env/bin/python -m celery -b amqp://... inspect active
        • tasks that this worker is able to execute are listed in:
          • if they are inside a Django project:
            • ...
          • if they are not inside a Django project:
            • ...
        • ...
      • client:
        • calls an async function
        • when called from a Django project
          • connects to broker specified in settings.py BROKER_URL
          • tasks that this client is able to call are listed in:
            • ...
          • ...

      		•  needed files


        broker
        worker mydir/celery.py
        client
  • First steps with Celery
    1. install a message broker (message transport)
      • RabbitMQ (AMQP)
        • RabbitMQ
        • Configuració / Setup (Django: same as in settings.py BROKER_URL)
          1. rabbitmqctl add_user myuser mypassword
          2. rabbitmqctl add_vhost myvhost
          3. rabbitmqctl set_permissions -p myvhost myuser ".*" ".*" ".*"
        • check settings:
          • rabbitmqctl list_users
          • rabbitmqctl list_vhosts
          • rabbitmqctl list_permissions -p myvhost
      • Amazon SQS
    2. install celery (dependencies automatically installed: pytz billiard kombu anyjson amqp)
      • pip install celery
    3. check that celery is working
  • command line





    		• examples
      celery -b <broker> -A <module> worker (start a single worker) -Q, --queues <queue1>,<queue2>
      --hostname=<node_name>@<host> (default: celery@<value_returned_by_hostname_command>)
      -E, --task-events (needed if we are intercepting events)
      -c <number_processes> (default: number of CPUs)
      multi (start several named workers)
      • start <node_name_1> [<node_name_2>, ...]
        • --pidfile=/var/run/celery/%n.pid
        • --logfile=/var/log/celery/%n%I.log
      • restart <node_name> [<node_name_2>, ...]
      • stop <node_name> [<node_name_2>, ...]
      • stopwait <node_name> [<node_name_2>, ...]
      		(usually called from celery.service with parameters set by variables defined in /etc/sysconfig/myceleryconfig, parsed with EnvironmentFile=-/etc/sysconfig/myceleryconfig)
      /path/to/python -m celery multi start mynodename -E -Q myqueue -A myapp --workdir=/path/to/myworking_dir --loglevel=DEBUG --logfile="/var/log/celery/%N.log"       --pidfile="/var/run/celery/%N.pid"
      will start number_process workers (hostname is also called "worker name" in Flower):
      • /path/to/python -m celery worker -E -Q myqueue -A myapp --loglevel=DEBUG --logfile=/var/log/celery/mynodename.log --pidfile=/var/run/celery/mynodename.pid --hostname=mynodename@...
      • /path/to/python -m celery worker -E -Q myqueue -A myapp --loglevel=DEBUG --logfile=/var/log/celery/mynodename.log --pidfile=/var/run/celery/mynodename.pid --hostname=mynodename@...
      • ...
      These workers will connect to the broker specified as Celery.broker in /path/to/myworkingdir/myapp/celery.py, on queue myqueue:
      • from __future__ import absolute_import, unicode_literals
        from celery import Celery

        app = Celery('myapp',
                     broker="amqp://myuser:mypassword@ip_address_of_rabbitmq_server:5672/myvhost",
                     backend='rpc://',
                     include=['myapp.tasks'])

        if __name__ == '__main__':
            app.start()
      Available tasks are registered in /path/to/myworkingdir/myapp/tasks.py (and referenced as myapp.task.task1 in Flower):
      • @shared_task(queue='myqueue')
        def task1(...):
            ...

        @shared_task(queue='myqueue', resultrepr_maxsize=4096, bind=True, acks_late=True)
        def task2(self, ...):
            ...


      		inspect
      • active
      • scheduled
      • reserved
      • revoked
      • registered
      • stats
      • query_task <task_uuid>
      		 --destination=celery@example.com
      • connect to broker and get a list of active tasks, for all workers:
        • celery -b amqp://<celery_user>:<celery_password>@<rabbitmq_server/><celery_vhost> inspect active

      		control
      • enable_events
      • disable_events
      • rate_limit tasks.add 10/m



      		events --dump

      		status
      • connect to broker and list active nodes:
        • celery -b amqp://<celery_user>:<celery_password>@<rabbitmq_server/><celery_vhost> status
        • Problems
          • Error: No nodes replied within time constraint.
            • ...

      		...




  		• worker client

    		celery.py
    		 tasks.py
    		 command line
    		 service usage


    		tasks.py
    • from celery import Celery

      app = Celery('tasks', broker='pyamqp://guest@localhost//')

      @app.task
      def add(x, y):
          return x + y
    		 celery -A tasks worker --loglevel=info
    • from tasks import add

      result = add.delay(4, 4)
      result.ready()
      result.successful()
      result.get()
      result.failed()
      res.state
      res.id



    		proj/rabbitmq.txt
    • amqp://<celery_user>:<celery_password>@<rabbitmq_server/><celery_vhost>
    proj/celery.py
    • from __future__ import absolute_import, unicode_literals
      from celery import Celery

      app = Celery(
          'proj',
           broker=open('rabbitmq.txt','r').read().strip(),
           backend='rpc://',
           include=['proj.tasks']
      )

      # Optional configuration, see the application user guide.
      app.conf.update(
          result_expires=3600,
      )

      if __name__ == '__main__':
          app.start()
    		 proj/__init__.py

    proj/tasks.py
    • from __future__ import absolute_import, unicode_literals
      from .celery import app

      @app.task
      def add(x, y):
          return x + y

      @app.task
      def mul(x, y):
          return x * y

      @app.task
      def xsum(numbers):
          return sum(numbers)
    		 (from proj parent dir)
    celery -A proj worker -l info
    		(from proj parent dir)
    • from proj.tasks import add

      result = add.delay(4, 4)

    		mydir/mymodule/rabbitmq.txt
    • amqp://<celery_user>:<celery_password>@<rabbitmq_server/><celery_vhost>
    mydir/mymodule/celery.py
    • from celery import Celery

      app = Celery(
          'mymodule',
          broker=      open('rabbitmq.txt','r').read().strip(),
          backend='rpc://',
          include=['mymodule.tasks']
      )

      if __name__ == '__main__':
          app.start()
    		 mydir/mymodule/__init__.py

    mydir/mymodule/tasks.py
    • from celery import shared_task

      @shared_task(queue='myqueue')
      def add(x,y):
          ...
    		python -m celery multi start mycelery_nodes -Q myqueue -A mymodule
    • will use mymodule/celery.py
    • will contact broker specified in app = Celery() and join the party
    • used queue will be the one specified in @shared_task
    • from myproject.tasks import add

      result = add.delay(...)

      result = add.apply_async(...)
    Django
    mysite/myproject/settings.py
    • # celery
      BROKER_URL = 'amqp://myuser:mypassword@localhost/myvhost'
    		mysite/myproject/celery.py
    • from __future__ import absolute_import
      import os
      from celery import Celery
      from django.conf import settings

      # set the default Django settings module for the 'celery' program.
      os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

      app = Celery('myproject')

      # Using a string here means the worker will not have to
      # pickle the object when using Windows.
      app.config_from_object('django.conf:settings')

      # access to mysite/*/tasks.py
      app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

      # access to mysite/myproject/tasks.py
      app.autodiscover_tasks(lambda: ('myproject',))

      @app.task(bind=True)
      def debug_task(self):
          print('Request: {0!r}'.format(self.request))
    		mysite/myproject/__init__.py
    • from __future__ import absolute_import

      # This will make sure the app is always imported when
      # Django starts so that shared_task will use this app.
      from .celery import app as celery_app
    mysite/myproject/tasks.py
    • from __future__ import absolute_import
      import logging
      from celery import shared_task

      logger = logging.getLogger(__name__)

      @shared_task
      def add(x, y):
          logger.info("adding...")
          return x + y
    mysite/myapp/tasks.py
    • from __future__ import absolute_import
      import logging
      from celery import shared_task

      logger = logging.getLogger(__name__)

      @shared_task
      def mul(x, y):
          logger.info("multiplying...")
          return x * y
    		 celery -A myproject worker -l info
    		e.g.: mysite/myapp/views.py
    • from myapp.tasks import mul

      result = mul.delay(3, 4)
  • Next steps
  • User Guide
  • Progrés / Progress
  • Problemes / Problems
  • ...

Exemples / Examples
  • Python Command Line Arguments Examples


    		• list of arguments
      		 number of arguments
      		 script
      		 argument






      sys
      		 import sys
      		 # including script
      sys.argv      		 # including script
      len(sys.argv)      		 sys.argv[0] sys.argv[1]
      sys.argv[2]
      ...
      getopt
      		 import sys, getopt


      		try:
          myopts, args = getopt.getopt(sys.argv[1:],"i:o:")
      except getopt.GetoptError as e:
          print (str(e))
          print("Usage: %s -i input -o output" % sys.argv[0])
          sys.exit(2)
       
      for o, a in myopts:
          if o == '-i':
              ifile=a
          elif o == '-o':
              ofile=a
      argparse



      		import argparse
      import textwrap

      #parser = argparse.ArgumentParser(description='ADD YOUR DESCRIPTION HERE')
      parser = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter,
                                       description=textwrap.dedent('''\
                                       First line
                                       and second line
                                       '''))
      parser.add_argument('-i','--input', help='Input file name', required=True)
      parser.add_argument('-t','--title', metavar='titol', help='Input file name', required=False)
      parser.add_argument('-n','--number', type=int, help='Input file name', required=True)
      parser.add_argument('first_parameter', help='...')

      parser.add_argument('parameter_1')
      parser.add_argument('parameter_2')
      parser.add_argument('--disable', dest='is_enabled', action='store_false', required=False)
      parser.add_argument('-t', '--title', metavar='titol',
                          help='title to be put in field s= in sdp file',
                          required=False)
         
      args = parser.parse_args()
      print(args)
      print("parameter_1: {0}".format(args.parameter_1))
      print("title: {0}".format(args.titol))
  • print
    • print ("nombre_args: %d" % nombre_args)
  • Fusió d'intervals / Merge intervals

IDE
  • Comparison of integrated development environments: Python (wp)
  • IDLE (wp)
  • PyDev (Eclipse)
    • Installation
      • Help -> Install new software -> Add ...:
        • PyDev - https://pydev.org/updates
      • PyDev Perspective
        • PyDev Package Explorer
          • Top Level Elements: Projects
    • Django
    • New project
      • workspace: ~/src/djcode/
      • project name: nom_projecte
      • will create (according to new directory layout in Django 1.4):
        • ~/src/djcode/
          • nom_projecte/
            • .project
              • <name>nom_projecte</name>
            • .pydevproject
            • sqlite.db
            • nom_projecte/
              • settings.py
              • ...
      • nom_projecte can be renamed to nom_projecte_pare (if you get the error "out of sync with file system", do a Refresh) (but DATABASES in settings.py is not updated):
        • ~/src/djcode/
          • nom_projecte_pare/
            • .project
              • <name>nom_projecte_pare</name>
            • .pydevproject
            • sqlite.db
            • nom_projecte/
              • settings.py
              • ...
      • les noves aplicacions (nom_projecte_pare -> Django -> create new app) es crearan dins de nom_projecte (de fet, dins del directori especificat a .pydevproject com a DJANGO_MANAGE_LOCATION)
    • virtualenv
      • Integrar Virtualenv con Eclipse (PyDev)
      • Pydev and virtualenv
      • passos / steps
        • general configuration:
          • Window -> Preferences -> PyDev -> Interpreter - Python: New...
            • Interpreter Name: python-PYTHON27
            • Interpreter Executable: /opt/PYTHON27/bin/python
        • on your project:
          • Properties -> PyDev - Interpreter/Grammar
            • Interpreter: python-PYTHON27
    • Problems
      • debugging suspends on caught exceptions ("VariableDoesNotExist: Failed lookup for key...") related to Django templates:
        • Solució / Solution
          • PyDev -> Manage exception breakpoints
            • Uncheck: "Suspend on django template render exceptions"
      • Unable to read repository at http://pydev.org/updates/content.xml Transport initialization error..
        • Solution:
          • rm ~/.eclipse
          • eclipse
      • New added library (e.g. by using pip) not detected
        • Solution:
          • Window -> Preferences -> PyDev -> Interpreter - Python Interpreter -> Remove -> AutoConfig
    • Existing code (1.3)
      • djcode/
        • mysite/
          • settings.py
          • mystite.db
          • polls/

GUI
  • tkinter

Frameworks

Google API

http://www.francescpinyol.cat/python.html
Primera versió: / First version: 24 VIII 2015
Darrera modificació: 31 d'agost de 2021 / Last update: 31st August 2021

Valid HTML 4.01!

Cap a casa / Back home