PythonSnips

From Custard
Jump to: navigation, search

Contents

Python Snips

PDB Crib

import pdb; pdb.set_trace()
python -m pdb code.py

p - print variable value
pp - pretty print
a - print input args of function
display j - display value as it changes
locals - show local variables

n - next
s - step into
c - continue
r - run until return
<Enter> - repeat previous command
l - print nearby code
l from,to - print code between lines
w - where am I + traceback
u - up the call stack
d - down the call stack

j 15 - Jump ahead to line 15
run - restart program

b 45 - set on line 45
b 45, x == 'abc' - set with condition
b - list current breakp­oints
b funcName - set breakpoint in function
b file.py:41 - set on line 41 of file.py
cl - clear all breakp­oints
cl 42 - clear breakpoint #42
tbreak ... - hit only once, syntax as b
disable 42 - turn off breakpoint #42
enable 42 - enable breakpoint #42
ignore 42 [N] - ignore bp#42 N times
condition 42 x=='abc' - set condition to bp (no condition = remove it)

alias - display / set aliases

Download Blob from Azure

    blob_service = BlobService(account_name, account_key)
    blob_list = blob_service.list_blobs(container)

    for blob in blob_list:
        blob_service.get_blob_to_path(containe, blob.name, os.path.join(target_dir, blob.name))

Timezone crazy

import time

time.time()                                          # 1461227070.572573
import datetime
datetime.datetime.now()                              # 2016-04-21 08:32:24.972434
datetime.datetime.utcnow()                           # 2016-04-21 08:36:19.027780

datetime.datetime.fromtimestamp(time.time()).tzname()

datetime.datetime.fromtimestamp(time.time())         # datetime.datetime(2016, 4, 21, 9, 44, 8, 831758)
str(datetime.datetime.fromtimestamp(time.time()))    # '2016-04-21 09:49:48.259503'

str(datetime.datetime.utcfromtimestamp(time.time())) # '2016-04-21 08:53:31.851301'
from django.utils import timezone

# settings.py
# USE_TZ = True
# TIME_ZONE = 'UTC'

timezone.now()                                       # 2016-04-21 08:32:40.285163+00:00

Writing Dict's to CSV

import csv

data = {
    "Calls": 100,
    "Call cost": 10,
    "Subtotal": 1000,
    "Total": 1000,
}

with open('totals.csv', 'w') as csvfile:
    fieldnames = data.keys()
    writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerow(data)
Total,Subtotal,Call cost,Calls
1000,1000,10,100

Python 3 Enums

from enum import IntEnum
class Poops(IntEnum):
    POOP = 0
    SCOOP = 1

poop = Poops.POOP
print(poop)        # Poops.POOP
print(str(poop))   # Poops.POOP
print(poop.name)   # POOP
print(int(poop))   # 0
print(poop.value)  # 0

Django serializer validation order

Validation is run in a specific, undocumented order

    Field deserialization called (serializer.to_internal_value and field.run_validators)
    serializer.validate_[field] is called for each field
    Serializer-level validators are called (serializer.run_validation followed by serializer.run_validators)
    serializer.validate is called

See: http://stackoverflow.com/questions/27591574/order-of-serializer-validation-in-django-rest-framework

Generating API from config using Swagger

Spin up VM and Docker image

git clone http://github.com/swagger-api/swagger-codegen.git
cd swagger-codegen
vagrant up
vagrant ssh
cd /vagrant
./run-in-docker.sh mvn package

Build Example

Run swagger-codegen in docker against petstore example json. The docker image mounts vagrant as /gen and puts generated files there.

./run-in-docker.sh generate -i http://petstore.swagger.io/v2/swagger.json -l python -o /gen/samples/test/
ls -al samples/test

Shell into Docker container:

docker run -it -w /gen -v /vagrant:/gen -v /home/vagrant/.m2/repository:/root/.m2/repository maven:3-jdk-7 bash

Ternery operator Python 3

[on_true] if [expression] else [on_false]

x, y = 50, 25
small = x if x < y else y

Dict comprehensions Python >2.7

        # Filter out None values
        params = {k: v for k, v in params.iteritems() if v is not None}

Standard Exceptions

BaseException
 +-- SystemExit
 +-- KeyboardInterrupt
 +-- GeneratorExit
 +-- Exception
      +-- StopIteration
      +-- StandardError
      |    +-- BufferError
      |    +-- ArithmeticError
      |    |    +-- FloatingPointError
      |    |    +-- OverflowError
      |    |    +-- ZeroDivisionError
      |    +-- AssertionError
      |    +-- AttributeError
      |    +-- EnvironmentError
      |    |    +-- IOError
      |    |    +-- OSError
      |    |         +-- WindowsError (Windows)
      |    |         +-- VMSError (VMS)
      |    +-- EOFError
      |    +-- ImportError
      |    +-- LookupError
      |    |    +-- IndexError
      |    |    +-- KeyError
      |    +-- MemoryError
      |    +-- NameError
      |    |    +-- UnboundLocalError
      |    +-- ReferenceError
      |    +-- RuntimeError
      |    |    +-- NotImplementedError
      |    +-- SyntaxError
      |    |    +-- IndentationError
      |    |         +-- TabError
      |    +-- SystemError
      |    +-- TypeError
      |    +-- ValueError
      |         +-- UnicodeError
      |              +-- UnicodeDecodeError
      |              +-- UnicodeEncodeError
      |              +-- UnicodeTranslateError
      +-- Warning
           +-- DeprecationWarning
           +-- PendingDeprecationWarning
           +-- RuntimeWarning
           +-- SyntaxWarning
           +-- UserWarning
           +-- FutureWarning
	   +-- ImportWarning
	   +-- UnicodeWarning
	   +-- BytesWarning

Rabbits

Adventures with RabbitMQ

  • sudo apt-get install rabbitmq-server
  • pip install pika
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

### Pub

channel.queue_declare(queue='pomegranate')
channel.basic_publish(
    exchange='',
    routing_key='pomegranate',
    body='I can haz pomegranatez!'
)

### Sub

def callback(ch, method, properties, body):
    print("%r" % body)

channel.basic_consume(
    callback,
    queue='pomegranate',
    no_ack=True
)

channel.start_consuming()

PyRes - Resque clone

  1. Module: my.module.hello
class Hello():
    queue = "Hello"

    @staticmethod
    def perform(thing):
        print("Hello %s" % thing)
        return True
  1. File: world.py
from pyres import ResQ
from my.module.hello import Hello
import logging

logging.basicConfig(level=logging.DEBUG)

ResQ().enqueue( Hello, "World" )
$ python world.py 
INFO:pyres:enqueued 'my.module.hello.Hello' job on queue Hello
DEBUG:pyres:job arguments: ('World',)

$ pyres_worker Hello
2015-12-03 16:10:09  5975 INFO     starting
2015-12-03 16:10:09  5975 INFO     Found job on Hello: (Job{Hello} | my.module.hello.Hello | [u'World'])
2015-12-03 16:10:09  5975 INFO     Forked 5982 at 2015-12-03 16:10:09.622648
2015-12-03 16:10:09  5982 INFO     Processing (Job{Hello} | my.module.hello.Hello | [u'World']) since 2015-12-03 16:10:09.623169
Hello World

Django

Install Django

Currently version 1.3

sudo easy_install django

or

apt-get install python-django

Test the install...

python
>>> import django
>>> print django.get_version()
1.3

Create a new project

django-admin.py startproject my_project

cd my_project

python manage.py runserver
python manage.py runserver 192.168.1.250:8000

Create an app under the project

python manage.py startapp my_app

edit urls.py

urlpatterns = patterns('',
    # Example:
    (r'^my_app/', 'my_project.my_app.views.index'),

add a view to my_app/views.py

from django.http import HttpResponse

def index(request):
    return HttpResponse("Index View")

Iterator

#!/usr/bin/python

class Things:

	def __init__(self) :
		self.data = ["Hello","World"]
		self.index = 0

	def __iter__(self) :
		return self

	def next(self) :
		if self.index<len(self.data) :
			self.index = self.index + 1
			return self.data[self.index-1]
		else:
			raise StopIteration

And a quick test...

things = Things()
for thing in things : 
	print thing

Bottle

A bit like Ruby's Sinatra but for Python. Cook up quick webapps and apis with minimal fuss..

#!/usr/bin/python

from bottle import route, post, get, run

# http://bottle.paws.de/docs/dev/tutorial.html#getting-started

class Bottletest:

        def __init__( self ):
                print( "Bonjiourno!" );

        @route('/')
        @route('/index.html')
        def index():
                return( "Index" )

        @route('/hello/:name')
        def hello( name ):
                return( "Hello %s" % name )

        @post('/login')
        def login():
                name     = request.forms.get('name')
                password = request.forms.get('pass')
                return( "name: %s pass: %s" % name,password )

        def run( self ):
                run(host='localhost', port=8123)

Sesame / SwiftOwlim

#!/usr/bin/python

from rdflib import ConjunctiveGraph
from rdflib import BNode, Literal
from SPARQLWrapper import SPARQLWrapper, JSON
import rdflib
from rdflib import Graph
import pickle
import pprint;

pp = pprint.PrettyPrinter( indent = 4 )

rdflib.plugin.register('sparql', rdflib.query.Processor,
                       'rdfextras.sparql.processor', 'Processor')
rdflib.plugin.register('sparql', rdflib.query.Result,
                       'rdfextras.sparql.query', 'SPARQLQueryResult')

url = "http://localhost:8080/openrdf-sesame/repositories/swiftowlimTest"

sparql = SPARQLWrapper( url )

graph = Graph()

#s = RDF::URI("http://pomegranate.heroku.com")
#p = RDF::DC.creator
o = Literal( "bruce" )

query= "SELECT * WHERE { ?s ?p ?o }"

sparql.setQuery( query );

result = sparql.query()
for triple in result:
	pp.pprint( triple )

Kamaelia

Kamaelia - Concurrency made useful, fun.

http://www.kamaelia.org/Home.html http://www.kamaelia.org/GettingStarted.html

Download Kamaelia

http://www.kamaelia.org/release/MonthlyReleases/Kamaelia-1.0.12.0.tar.gz

Install Kamaelia

cd Kamaelia-1.0.12.0/
sudo python setup.py install
Personal tools