Python

Índex

General

Python 3

Python

  • The Pyhton tutorial
  • PEP
  • Estil / Style
  • Python (Raspberry Pi)
  • PyPI: the Python package index
  • Python Documentation contents
    • The Python standard library
      • 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
      • Test unitari / Unit test

        • pytest unittest Django (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
        • pytest
        • unittest
        • Django tests
        • mock
        • ...
      • CGI
      • URL
      • 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
      • mime types
        • 18.7 mimetypes
          • import mimetypes
          • guessed_mime_type = mimetypes.guess_type()
          • 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
        • 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
      • 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
    • 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)
      • 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
            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)

            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:
            #from patchwork import files
            #files.append(c, '/etc/sysconfig/spawn-fcgi', text, sudo=True)
        • 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 ...
      • 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:
    • /usr/local/lib/python2.7/dist-packages/...
    • 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

Biblioteques / Libraries

          • 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}

                        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

Celery

  • Celery with Django
  • Arquitectura / Architecture
  • 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 <queue1>,<queue2>
      --hostname=<node_name>@<host> (default: celery@<value_returned_by_hostname_command>)

      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>, ...]
      -c <number_processes> (default: number of CPUs)
      (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

      ...


  • celery.py
    tasks.py
    worker
    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)
    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
  • ...

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
      parser = argparse.ArgumentParser(description='ADD YOUR DESCRIPTION HERE')
      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
    • 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ó: 22 d'octubre de 2020 / Last update: 22th October 2020

Valid HTML 4.01!

Cap a casa / Back home