Python
Índex
- General
- Python 3
- Python
- Documentació / Documentation
- Dades / Data (object, list, dict, tuple)
- Test unitari / Unit test
- CGI
- URL
- Expressions regulars / Regular expressions
- JSON
- mime-types
- XML
- Loops
- Sequence types
- Mapping types (dict)
- Data / Date
- Adreces de xarxa / Network addresses
- Random
- Unicode
- Eines
- Documentació / Documentation
- Biblioteques /
Libraries
- Imatges / Images
- Criptografia / Cryptography
- Xarxa / Network
- Processos / Processes
- SSH
- SVG
- FTP
- HTTP
- PyGObject
- Parsing
- Exemples / Examples
- Arguments (argparse)
- IDE
- GUI
- Frameworks
- Google API
General
Python 3
- Porting Python 2 Code to Python 3
- check good coverage of tests: use coverage.py
- make your code compatible with python 2 and python 3: use futurize (modernize)
- clean your code: use pylint
- check whether dependent modules can be ported to python 3: use caniusepython3
- test your code under several versions of python: use tox
- Supporting Python 3: An in-depth guide
- Cheat Sheet: Writing Python 2-3 compatible code
- 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')
- from (row) -> to (column)
-
Python
- The Pyhton tutorial
- PEP
- Estil / Style
- PEP-8: Style Guide for Python Code
- The
Black code style
- Black Playground (online formatter)
- Fitxer / File
#!/usr/bin/env python3
# -*- coding: utf-8 -*-- ...
- Python
(Raspberry
Pi)
- Python
2 vs Python 3
- ipython
- Python
2 vs Python 3
- 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
- Eclipse
- Ús / Usage
- Basic patterns and examples
- executa un test en concret
pytest toto.py -k test_primer
- mostra tots els print (si no, només es
mostraran dels tests que fallin)
pytest -s ...
- ...
- unittest
- Django tests
- mock
- ...
-
- CGI
- 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()
- quote()
- quote_plus()
- unquote()
- unquote_plus()
- urlencode()
- pathname2url()
- url2pathname()
- getproxies()
urllib.parse
- urlparse()
- parse_qs()
- parse_qsl()
- urlunparse()
- urlsplit()
- urlunsplit()
- urljoin()
- urldefrag()
- unwrap()
- ...
- urlquote()
- urlquote_plus()
- quote_from_bytes()
- unquote()
- unquote_plus()
- unquote_to_bytes()
- urlencode()
urllib.robotparser - python 2: urlparse
from urlparse import urlparse
- python 3: urllib.parse
from urllib.parse import urlparse
o = urlparse('http://www.cwi.nl:80/%7Eguido/Python.html')
oo = urlparse('s3://bucket/path/to/file')
o
- 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
- unquote
- convert: /path/to/toto_%C3%A0.mp4 -> /path/to/toto_à.mp4
path_original = u'/path/to/toto_%C3%A0.mp4'
path_converted = urllib.unquote(path_original.encode()).decode('utf-8')
-
- urljoin
- How
to join components of a path when you are
constructing a URL in Python
def urljoin(*args):
"""
Joins given arguments into a url. Trailing but not leading slashes are
stripped for each argument.
"""
return "/".join(map(lambda x: str(x).rstrip('/'), args))
- How
to join components of a path when you are
constructing a URL in Python
- 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
- Mageia
- 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")
- Dependències / dependencies
- 18.7
mimetypes
- XML
- 19 Structured Markup Processing Tools
- xml.etree.ElementTree
–
XML Manipulation API
- Creating XML documents
- Use
ElementTree's findall() to match text of
elements
root.findall('.//planet[name="Kepler"]')
- defusedxml
- lxml
- lxml API
- Write xml file using lxml library in Python
- lxml question -- creating an etree.Element attribute with ':' in the name
- Python: adding namespaces in lxml
- Example to create a document:
from lxml import etree
# root
arrel = etree.Element('root')
# subelements
# print string
print( etree.tostring(arrel, encoding='utf-8', xml_declaration=True, pretty_print=True).decode('utf-8') )
from lxml import etree as ET
# root
NS = "http://www.w3.org/2001/XMLSchema-instance"
location_attribute = '{%s}noNameSpaceSchemaLocation' % NS
appearancetable = ET.Element('appearancetable', attrib={location_attribute: "http://jmri.org/xml/schema/appearancetable.xsd"})
tree = ET.ElementTree(appearancetable)
# subelements
father = ET.SubElement(appearancetable, 'father')
father.set('myattribute', 'myvalue')
father.text = 'text inside element'
# write to file
tree.write(xml_appearance_path, pretty_print=True, xml_declaration=True, encoding="utf-8")
- Loops
- compact
["{}-{}".format(a,b) for a,b in ...]
- compact
- 5.6.
Sequence
Types — str, unicode, list, tuple, bytearray, buffer,
xrange
- Llistes / Lists
- Dades / Data
- Sort
- Unique (remove repeated elements)
my_list_without_repeated_elements = list(set(my_list_with_repeated_elements))
- Convert string to list
- Search
- Python list of dictionaries search
[f for f in l if f['type']=='video' and f['format']=='hls']
- Current, next
- Iterate
a list as pair (current, next) in Python
for current, next in zip(the_list, the_list[1:]):
- Iterate
a list as pair (current, next) in Python
- Llistes / Lists
- 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)
- Pretty print (not recursive)
- 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, '...')
- now in UTC and ISO-8601 format
- 8.1.7
strftime()
and strptime() Bahavior
- 15.3 time - Time access and conversions
- minimum and maximum datetime (infinite)
unaware_maximum = datetime.datetime.maximport pytz
aware_maximum = datetime.datetime.max.replace(tzinfo=pytz.UTC)
- other libraries
- python-dateutil (based on mxDateTime)
- add datetime and time
- event.start + datetime.timedelta(hours=event.duration.hour,minutes=event.duration.minute,seconds=event.duration.second)
- difference between two times
- Difference between two time intervals in Python
- Issue 3250: datetime.time does not support arithmetic
- python-dateutil
- python time arithmetic with datetime?
duration = end - begin
# to also take account days (instead of just .seconds):
duration_in_seconds = duration.total_seconds()
- Date in Django
- Periods
- Generate
a list of datetimes between an interval in
python
def perdelta(start, end, delta):
curr = start
while curr < end:
yield curr, min(curr+delta, end)
curr += delta
- Generate
a list of datetimes between an interval in
python
- Fusos horaris / Timezones
import pytz
timezone = 'Europe/Andorra'
my_date.astimezone(pytz.timezone(timezone)).strftime('%H:%M:%S %Z')
- 8.1 datetime
- Adreces de xarxa / Network
addresses
- adreça IP pròpia / own IP address
- example:
import socket
socket.gethostbyname_ex(socket.gethostname())[2][0]
- Problemes / Problems
socket.gaierror: [Errno -2] Name or service not known- Why am I getting socket.gaierror: [Errno -2] from Python HTTPLib
- Solució / Solution:
import socketsocket.gethostname()- reesult must appear in /etc/hosts
- example:
- 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')
-
- adreça IP pròpia / own IP address
- 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)))
- type
- The
Python standard library
- 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(...)
- do not forget trailing 'u'
- 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)
- Use format instead of '%'
- 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
- Install virtualenv
- Python 3
- Mageia
urpmi python3-virtualenvurpmi lib64python3-devel
- CentOS
sudo yum install python36 python36-pip python36-devel python34-virtualenvvirtualenv-3 --python python36 env
- Mageia
- Python 2
- Mageia
urpmi python-pkg-resources python-virtualenv- Bug on Mageia 3 (makes "pip install pil"
fail)
- Bug 11283
- Python headers included are bungled,
probably due to multi-arch issues
- replace /usr/lib/python2.7/site-packages/virtualenv.py by virtualenv.py
- run again:
virtualenv ...
- Bug 11283
- Python headers included are bungled,
probably due to multi-arch issues
- CentOS
yum install python-virtualenv- option 1: install from SCL
- Setting up Django and Python 2.7 on Red Hat Enterprise 6 the easy way
scl -lscl enable python27 bash
- other options:
virtualenv -p python27 /opt/PYTHON27source /opt/PYTHON27/bin/activate- Problems:
- libpython2.7.so.1.0: cannot open
shared object file: no such file or
directory
- Solution:
scl enable python27 bash
- Solution:
- libpython2.7.so.1.0: cannot open
shared object file: no such file or
directory
- Ubuntu
sudo apt-get install python-virtualenv
- MSWindows
- pip install virtualenv
- Mageia
- Python 3
- Utilització / Usage
- Python 3
- Mageia
virtualenv env
- CentOS
- ...
- Mageia
- Python 2
- virtualenv [--distribute] /opt/PYTHON27
- force Python 2.6
virtualenv -p python26 /opt/PYTHON26
- MSWindows
virtualenv.exe c:\tmp\p27c:\tmp\p27\Scripts\activate...deactivate
source /opt/PYTHON27/bin/activate(/opt/PYTHON27)...$ ...deactivate
- Python 3
- virtualenv in PyDev
- virtualenv in Django
- mod_wsgi
- Django
- mod_wsgi
- CentOS
- Python 2.7 en CentOS 6 con mod_wsgi
- Please create a mod_wsgi build for the python 2.7 SCL environment on RHEL 6
- Using RHSCL: Django on Python 3 with PostgreSQL
- compilation:
- wget ...
- ./configure --with-python=/opt/PYTHON27/bin/python
- make
- make install
- config:
- conf.d/wsgi.conf
- LoadModule wsgi_module modules/mod_wsgi.so
- conf.d/wsgi.conf
- CentOS
- Install virtualenv
- pip
(*) (package manager) (also installed
when virtualenv
is installed)
- Installation of pip itself
- From distribution
- CentOS
sudo yum install python-pip
- CentOS
- From source
- download it:
- install it:
# python setup.py install
- From distribution
Installation of packagespip 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 -Upackage_name
- Problems
error fatal: Python.h: El fitxer o directori no existeix- Solució / Solution
- Python 2.7
- Mageia
urpmi lib64python-devel
- Mageia
- Python 3
- Mageia
urpmi lib64python3-devel
- Mageia
- Python 2.7
- Solució / Solution
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.crtsudo 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
- Solució / Solution
Error: pg_config executable not found.- Solució / Solution
- Install postgresql devel
- Mageia
urpmi postgresql9.4-devel
- Mageia
- Install postgresql devel
- Solució / Solution
- list of installed packages (it gives the version
number):
pip freeze
- Installation of pip itself
- easy_install
urpmi python-setuptoolssudo apt-get install python-setuptools
- Invoke
- split from fabric
- 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
- Documentation (1.10)
- Overview and tutorial
- fab options and arguments
- Execution model
- Operations
- fabric.operations.get
- fabric.operations.local
- fabric.operations.open_shell
- fabric.operations.prompt
- fabric.operations.put
- put a file
with given permissions
put('remote_server.pem', '~/.ssh/id_remote_server.pem', mode=int('600', 8))
- put a file
with given permissions
- fabric.operations.reboot
- fabric.operations.require
- fabric.operations.run
- fabric.operations.sudo
- The environment dictionary, env
- ssh config
- Tutorial
- 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()
- fabfile.py
- 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
- 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('...')
- ignore
- 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
- fabfile.py
- 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)
- run remotelly and get the value
- crontab
- fab_deploy.crontab (django-fab-deploy)
env.sudo('crontab -l | { cat; echo "0 0 * * * some entry"; } | 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
- rsync
exclude according to .gitignore &
.hgignore & svn:ignore like
--filter=:C
from fabric.contrib.project import rsync_project
rsync_project(local_dir='.', remote_dir='/var/www', exclude=('.git','tmp',) , extra_opts="--filter=':- .gitignore'")
- rsync
exclude according to .gitignore &
.hgignore & svn:ignore like
--filter=:C
- 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)
- append a line
- sed
- single quotes in sed
- Re: [Fab-user] Problems escaping double and single quotes within run()
env.run(r"sed -i.bak -r 's/environment = '\''DEVEL'\''/environment = '\''PRO'\''/g' config/status.php")
- 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')
- single quotes in sed
- put
- 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')
- certificate for https connections (curl will not need to
specify --cacert)
- 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))
- fabfile.py
- 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')
- fabfile.py
- 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')
- fabfile.py
- 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) )
- fabfile.py
- 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")
- fabfile.py
- Instal·lació / Installation
- Empaquetament / Packaging
- virtualenv
- Multiple platforms
- How can I find the current OS in Python? [duplicate]
import platform
platform.platform()
platform.system()- Popen
- Processos / Processes
- Python: Howto launch a full
process not a child process and retrieve the PID
- DETACHED_PROCESS, CREATE_NEW_PROCESS_GROUP
- From Apache / WSGI
- Fork process to execute executables through python/wsgi under Apache/Windows (movie without window; only audio)
- How
to
start daemon process from python on windows?
- subprocess.STARTUPINFO()
- Run
background
process from PHP on Windows with window
- wintty
- PHP
execute
.exe file on the server
- psexec.exe (PSTools)
- I wish to execute .exe
file from my php program. I am using Window 7,
WAMP.
- psexec
- Kill a process
- Restart computer
- restart local computer from python
- Exemple / Example:
import platform
def restart_computer():
operating_system = platform.system()
if operating_system == 'Linux':
os.system('reboot now')
print "rebooting system"
elif operating_system == 'Windows':
import win32api
win32api.InitiateSystemShutdown()
- 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
- python
- set path:
- python
import sys
sys.path.append("/my/path")
- Django:
- /etc/apache2/mods-available/wsgi.conf
- WSGIPythonPath ...
- /etc/apache2/mods-available/wsgi.conf
- python
- 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
- Problemes / Problems
'ascii' codec can't encode character ... in position ...: ordinal not in range(128)- "UnicodeEncodeError: 'ascii' codec can't encode character" on peru sync #136
- Changing default encoding of Python?
- PrintFails
- Solució / Solution
- force utf8 mode when sys.stdout.encoding is ASCII #138
# https://github.com/buildinspace/peru/issues/136
# https://github.com/buildinspace/peru/pull/138/commits/8d368811dd6ae868f40f9c74fc87b0cbb8a9e58e
if sys.stdout.encoding == 'ANSI_X3.4-1968':
sys.stdout = open(sys.stdout.fileno(), mode='w', encoding='utf8', buffering=1)
sys.stderr = open(sys.stderr.fileno(), mode='w', encoding='utf8', buffering=1)
print("sys.stdout.encoding: {}".format(sys.stdout.encoding))
- Problemes / Problems
Biblioteques / Libraries
- Useful
modules
- Dades / Data
- Bases de dades / Databases
- sqlite
- Problemes / Problems
- How can I dump my sqlite3 database to SQL in UTF8 with django?
- Issue 15109: sqlite3.Connection.iterdump() dies with encoding exception (Python)
/usr/lib/python2.7/sqlite3/dump.py
- Problemes / Problems
- MySQL
pip install MySQL-pythonpip install mysql-python- Problems
EnvironmentError: mysql_config not found- Solution
- Mageia
urpmi lib64mariadb-devel
- CentOS
yum install mariadb-devel
- Ubuntu
sudo apt-get install libmariadbd-dev libpython-dev libssl-dev
- Mageia
- Solution
- /usr/include/mysql/my_config.h:654:2:
error:
#error <my_config.h> MUST be included
first!
- Solution
- /usr/include/mysql/my_config.h:654
#ifdef __GLIBC__
/*#error <my_config.h> MUST be included first!*/
#endif
- /usr/include/mysql/my_config.h:654
- Solution
- sqlite
- Imatges / Images
- python-imaging
- Pillow
- documentation
- Install
- requisites
- CentOS
yum install libjpeg-turbo-devel
- CentOS
- requisites
- Pillow
- PIL
(use Pillow
instead)
- pending deprecation in Django 1.6 (deprecated in Django 1.8)
- ImageIO
- Animated GIF
- python-imaging
- Criptografia / Cryptography
- hashlib
- base64
- url
safe (wp)
base64.urlsafe_b64encode(...)(to match Android URL_SAFE)- urlsafe_b64encode always ends in '=' ?:
- ? ((to match Android NO_PADDING)
- url
safe (wp)
- OpenSSL
- M2Crypto
- installation
urpmi python-m2crypto
- HOWTO: Programming S/MIME in Python with M2Crypto
- Examples
- installation
- pyOpenSSL
urpmi python-OpenSSL- Python OpenSSL Manual
/usr/share/doc/python-OpenSSL/examples/
- M2Crypto
- Autenticació
/ Authentication
- Django
- Twitter libraries
- OAuth
- rauth
- Twitter example
- Facebook
example
- cd src
- git clone https://github.com/litl/rauth.git
- cd rauth/examples/facebook
- virtualenv env
- source env/bin/activate
- pip install rauth flask flask-sqlalchemy
- python
- from facebook import db
- db.create_all()
- python facebook.py
- http://127.0.0.1:5000/
- Reuse tokens
- rauth
- Xarxa / Network
- Twisted
- stomp.py
- Problemes / Problems
UnicodeEncodeError: 'ascii' codec can't encode character ...- Solution
- /usr/local/lib/python2.7/dist-packages/stomp/protocol.py
#body = encode(body)
- /usr/local/lib/python2.7/dist-packages/stomp/protocol.py
- Solution
- Problemes / Problems
- netcat / nc
- 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
- 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))
- import
shlex, subprocess
- standard
- Problemes / Problems
- Unicode was not supported by shlex split
# https://stackoverflow.com/questions/14218992/shlex-split-still-not-supporting-unicode#answer-14219159
#args = shlex.split(command_line)
args = map(lambda s: s.decode('utf-8'), shlex.split(command_line.encode('utf-8')))[Errno 12] Cannot allocate memory
- 17.1 subprocess (2)
(3)
- 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
- Paramiko
- 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
apt-get install python-rsvg
- Mageia
- Ú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)
...
- upload local or remote file to ftp:
- libftp
(2.7)
- HTTP / HTTPS
- python-requests
- Instal·lació / Installation
pip install requests- JWT
pip install requests-jwt- JSON Web Token (JWT) Authorization for Python's Requests (requests-jwt)
- Ús / Usage
- get
import requests
from pprint import pprint
url = ...
r = requests.get(url)
pprint( r.json() )
- get
- Instal·lació / Installation
- python-requests
- 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['...'] = '...'
withopen('/tmp/toto.png', 'rb') as field_1_file:
files = {'field_1': field_1_file}
r = requests.post(url, payload, headers=headers, files=files)
- models.py
- HTTPS
- SSL Cert Verification
r = requests.get(..., verify='certificate_file.crt')
- 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
- Solució / Solution
- 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))
- Solució / Solution: explicit conversion to json,
with json header
- empty request.DATA when receiving a put/patch:
- 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()
- Exemple HHTPS / HTTPS Example
- 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-develapt-get ...
- dependències / dependencies:
urpmi [python-gobject] python3-gobject3yum install python34-gobjectapt-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
urpmi lib64gst-gir1.0 lib64ges-gir1.0 lib64gstwebrtc-gir1.0 ...yum install ...apt-get install gir1.x-gst... gir1.x-ges...- compile GStreamer with PyGobject enabled
- ValueError: Namespace Gst not available
- ValueError: Namespace GstWebRTC not available
...
- Instal·lació /
Installation
- Sistema /
System
- CentOS
sudo yum install python36-gobject
- Mageia
urpmi ...
- CentOS
- 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
- Sistema /
System
- 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
- standard location:
- 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
- Pyparsing
- ...
- Pyparsing
- Fulls de càlcul / Spreadsheet
Celery
- Celery with Django
- Arquitectura / Architecture
- First
steps
with Celery
- install a message broker
(message transport)
- RabbitMQ
(AMQP)
- RabbitMQ
- Configuració / Setup (Django:
same as in settings.py BROKER_URL)
rabbitmqctl add_user myuser mypasswordrabbitmqctl add_vhost myvhostrabbitmqctl set_permissions -p myvhost myuser ".*" ".*" ".*"
- check settings:
rabbitmqctl list_usersrabbitmqctl list_vhostsrabbitmqctl list_permissions -p myvhost
- Amazon SQS
- RabbitMQ
(AMQP)
- install celery (dependencies automatically installed: pytz
billiard kombu anyjson amqp)
pip install celery
- check that celery is working
- install a message broker
(message transport)
- 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@...
- ...
- 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()
- @shared_task(queue='myqueue')
def task1(...):
...
@shared_task(queue='myqueue', resultrepr_maxsize=4096, bind=True, acks_late=True)
def task2(self, ...):
...
inspectactive- 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
controlenable_eventsdisable_eventsrate_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>
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
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
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
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
- Calling tasks
- Canvas:
designing workflows
- Signatures
- add.signature(2,2)
- add.s(2,2)
- Primitives
- group
- chain
- chord
- map
- starmap
- chunks
- Signatures
- Routing
- Remote control
- Timezone
- Optimization
- User
Guide
- Application
- Main name
- app = Celery('tasks')
- app.main
- @app.task
def add(x,y):
return x+y - add.name
- Configuration
- app.conf
- app.conf.update({...})
- app.config_from_object(...)
- app.config_from_envvar(...)
- app.conf.humanize(...)
- Laziness
- add.__evaluated__()
- Breaking the chain
- Abstract tasks
- Main name
- Tasks
- Basics
- ...
- Monitoring and Management Guide
- ...
- Application
- Progrés / Progress
- Problemes / Problems
- /var/log/messages
... python: File "...lib/python2.7/site-packages/amqp/transport.py", line 438, in _read
... python: s = recv(n - len(rbuf))
... python: socket.error: [Errno 104] Connection reset by peer - celery raise error: [Errno 104] Connection reset by peer after started #4867
- Received error "[Errno 104] Connection reset by peer" after 180 seconds when waiting on task result #5358
- Solució / Solution:
- ? implement retries
- using RabbitMQ:
- ...
- using Redis:
- ...
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
- Help -> Install new software -> Add ...:
- 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
- ...
- .project
- nom_projecte/
- ~/src/djcode/
- 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
- ...
- .project
- nom_projecte_pare/
- ~/src/djcode/
- 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
- Window -> Preferences -> PyDev ->
Interpreter - Python: New...
- on your project:
- Properties -> PyDev - Interpreter/Grammar
- Interpreter: python-PYTHON27
- Properties -> PyDev - Interpreter/Grammar
- general configuration:
- 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"
- PyDev -> Manage exception breakpoints
- Solució / Solution
- Unable to read repository at
http://pydev.org/updates/content.xml Transport
initialization error..
- Solution:
- rm ~/.eclipse
- eclipse
- Solution:
- New added library (e.g. by using pip) not detected
- Solution:
- Window -> Preferences -> PyDev -> Interpreter - Python Interpreter -> Remove -> AutoConfig
- Solution:
- debugging suspends on caught exceptions ("
- Existing code (1.3)
- djcode/
- mysite/
- settings.py
- mystite.db
- polls/
- mysite/
- djcode/
- Installation
GUI
- tkinter
Frameworks
- Django
- Flask
- Dependencies
- Werkzeug
- Jinja 2 (templates)
- SQLAlchemy (database toolkit for Python)
- Logs
- Cache
- Dependencies
- REST
Google API
http://www.francescpinyol.cat/python.html
Primera versió: / First version: 24 VIII 2015
Darrera modificació: 25 de gener de 2021 / Last update: 25th
January 2021
Cap a casa / Back home