Apr 02-04, 2018 Vikrant Patil
These notes are available online at http://notes.pipal.in/2018/vmware-advanced-apr
© Pipal Academy LLP
We will be using anaconda (python 3) distribution for this training. it can be downloaded from
class A:
pass
class B(A):
pass
class C:
pass
class D(A,C):
pass
#class E(A,B):
# pass
D.__module__
D.__mro__
You want to create a class whose instances are created millions of times...you want to makee sure memory efficiecy
class OptimizedDate:
__slots__ = ['day','month','year']
def __init__(self, d, m ,y):
self.day = d
self.month = m
self.year = y
def __repr__(self):
return "OptimizedDate({},{},{})".format(self.day, self.month, self.year)
o = OptimizedDate(1,1,2018)
o
o.x = 0
o.day = 2
o.day = [1,2,3,4]
o
functions decorators get function as argument and they return function as return value. class decorator will recieve class as an argument and return class
def debug(cls):
print(cls)
return cls
@debug
class Foo: ## Foo = debug(Foo)
def __init__(self):
pass
def method1(self):
pass
def method2(self):
pass
Foo()
def debug(cls):
def logger(f):
def wrapper(*args, **kwargs):
print("Begning..", f.__qualname__)
return f(*args, **kwargs)
return wrapper
for name, value in vars(cls).items():
if callable(value):
setattr(cls, name, logger(value))
return cls
@debug
class Foo: ## Foo = debug(Foo)
def __init__(self):
pass
def method1(self):
pass
def method2(self):
pass
f = Foo()
f.method1()
f.method2()
class A:
def smethod():
print("A: Hello from smethod")
class B:
@staticmethod
def smethod():
print("B: Hello from smethod")
A
B
A.smethod()
B.smethod()
a = A()
a.smethod()
b = B()
b.smethod()
import datetime
class Mydatetime(datetime.datetime):
pass
datetime.datetime.now()
Mydatetime.now()
class A:
@staticmethod
def new():
return A()
class B(A):
pass
B.new()
A.new()
class A:
@classmethod
def new(cls, *args):
return cls()
class B(A):
pass
B.new()
class A:
def f1():
pass
def f2(self):
pass
a = A()
a.f1()
A.f1
A.f2
A.f2(a) ## >> a.f2()
A.f1()
a.f1() ## >>> A.f1(a)
class Person:
def __init__(self, name, surname, email):
self._name = name
self._surname = surname
self._email = email
@property
def name(self):
return " ".join([self._name, self._surname])
@name.setter
def name(self, value):
self._name, self._surname = value.split(" ")
@name.deleter
def name(self):
raise Exception("Can not delete")
p = Person("David", "Beazly", "david@python.org")
p.name
p.name = "Anand Chitipothu"
p.name
del p.name
type(p.name)
# Descriptor attribute for an integer type-checked attribute
# recepie taken from python cookbook, by brian k jones and david beazley
class Integer:
def __init__(self, name):
self.name = name
def __get__(self, instance, cls):
print("__get__ from", self)
if instance is None:
return self
elif self.name not in instance.__dict__:
raise AttributeError("No such attribute")
return instance.__dict__[self.name]
def __set__(self, instance, value):
print("__set__ from ", self)
if not isinstance(value, int):
raise TypeError('Expected an int')
instance.__dict__[self.name] = value
def __delete__(self, instance):
print("__del__ from", self)
del instance.__dict__[self.name]
def __str__(self):
return "Integer<{0.name!s}>".format(self)
class Point:
x = Integer('x')
y = Integer('y')
def __init__(self, x, y):
self.x = x
self.y = y
p = Point(2,3)
p.x
type(p.x)
del p.x
p.x
problem
class Person:
def __init__(self, name, surname, email):
self._name = name
self._surname = surname
self._email = email
@property
def name(self):
return " ".join([self._name, self._surname])
@name.setter
def name(self, value):
self._name, self._surname = value.split(" ")
@name.deleter
def name(self):
raise Exception("Can not delete")
class my_property:
def __init__(self, f):
self.function = f
def __get__(self, instance, cls):
if not instance:
return self
else:
return self.function(instance)
def __set__(self, instance, value):
return self.function(instance, value)
def __delete__(self, instance):
return self.function(instance)
@classmethod
def setter(cls, f):
return cls(f)
@classmethod
def deleter(cls, f):
return cls(f)
class P:
def __init__(self, name):
self._name = name
@my_property
def email(self):
return "@".join([self._name, "vmware.com"])
p = P("xyz")
p.email
# Descriptor attribute for an integer type-checked attribute
# recepie taken from python cookbook, by brian k jones and david beazley
class Integer:
def __init__(self, name):
self.name = name
def __get__(self, instance, cls):
print("__get__ from", self,"*", instance,"*", cls)
if instance is None:
return self
elif self.name not in instance.__dict__:
raise AttributeError("No such attribute")
return instance.__dict__[self.name]
def __set__(self, instance, value):
print("__set__ from ", self, instance, value)
if not isinstance(value, int):
raise TypeError('Expected an int')
instance.__dict__[self.name] = value
def __delete__(self, instance):
print("__del__ from", self, instance)
del instance.__dict__[self.name]
def __str__(self):
return "Integer<{0.name!s}>".format(self)
class Point:
x = Integer('x')
y = Integer('y')
def __init__(self, x, y):
self.x = x
self.y = y
point = Point(1,2)
Point.x
with open("data.csv") as f:
print(f.read())
%%file mytcpserver.py
import socketserver
import socket
import sys
class MyTCPHandler(socketserver.BaseRequestHandler):
def handle(self):
data = self.request.recv(2048).strip()
print ("%s wrote: " % self.client_address[0])
print (data)
if data=="close":
self.request.close()
self.server.shutdown()
else:
self.request.send(data.upper())
def start_server():
HOST, PORT = "127.0.0.1", 3456
server = socketserver.TCPServer((HOST, PORT), MyTCPHandler)
server.serve_forever()
def send_client_msg():
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(("localhost",3456))
s.send("This is junk data KJHJKASGHD jhG ASDGH AHSGDJASGH DFJKAGH:OP{IO "*10)
received = s.recv(1024)
s.close()
print ("Received: " + received)
if __name__ == "__main__":
start_server()
from socket import socket, AF_INET, SOCK_STREAM #from python cookbook, 3rd edition
class LazyConnection:
def __init__(self, address, family=AF_INET, type=SOCK_STREAM):
self.address = address
self.family = AF_INET
self.type = SOCK_STREAM
self.sock = None
def __enter__(self):
print("__enter__")
if self.sock is not None:
raise RuntimeError("Already connected")
self.sock = socket(self.family, self.type)
self.sock.connect(self.address)
return self.sock
def __exit__(self, exception_type, exception_value, traceback):
print("__exit__")
self.sock.close()
self.sock = None
import time
address = ("127.0.0.1", 3456)
with LazyConnection(address) as conn:## here enter will be called
time.sleep(3)
conn.send(b"Hello")
msg = conn.recv(1024)
print(msg.decode())
#here exit will be called
import threading
import time
def tick(n):
for i in range(n):
print("Tick", i)
time.sleep(1)
thread = threading.Thread(target=tick, args=(10,))
thread.start()
import time
def tick(n):
for i in range(n):
print("Tick", i)
time.sleep(2)
thread = threading.Thread(target=tick, args=(10,))
thread.start()
thread.join() # waits till thread finishes
2 + 3
%%file thread.py
import threading
import time
def tick(n):
for i in range(n):
print("Tick", i)
time.sleep(2)
if __name__ == "__main__":
thread = threading.Thread(target=tick, args=(5,))
thread.start()
thread.join() # waits till thread finishes
print("Thread is over!")
!python thread.py
%%file tdownload.py
import threading
import requests
import sys
import time
def wait(*args):
time.sleep(1)
def download(url):
requests.get(url)
#print("Finished!")
def executetask(concurrency, tasksize=10):
argsq = [("http://httpbin.org/html",) for i in range(tasksize)]
while argsq:
numt = concurrency if len(argsq)>concurrency else len(argsq)
threads = [threading.Thread(target=wait, args=argsq.pop()) for i in range(numt)]
for t in threads:
t.start()
for t in threads:
t.join()
if __name__ == "__main__":
executetask(int(sys.argv[1]))
!python tdownload.py 1
!python tdownload.py 2
!time -p python tdownload.py 1
!time -p python tdownload.py 2
!time -p python tdownload.py 4
%%file tpdownload.py
from multiprocessing.pool import ThreadPool
import requests
import sys
import time
def wait(*args):
time.sleep(1)
def download(url):
requests.get(url)
#print("Finished!")
def executetask(concurrency, tasksize=10):
argsq = [("http://httpbin.org/html",) for i in range(tasksize)]
pool = ThreadPool(concurrency)
r = pool.map(wait, argsq)
print(r)
if __name__ == "__main__":
executetask(int(sys.argv[1]))
!time -p python tpdownload.py 1
!time -p python tpdownload.py 2
!time -p python tpdownload.py 4
%%file tcpu.py
from multiprocessing.pool import ThreadPool
import requests
import sys
import time
def saxpy(*args):
s = 0
for i in range(1000):
for j in range(10000):
s += i*j*1.0
return s
def executetask(concurrency, tasksize=10):
argsq = [("http://httpbin.org/html",) for i in range(tasksize)]
pool = ThreadPool(concurrency)
r = pool.map(saxpy, argsq)
print(r)
if __name__ == "__main__":
executetask(int(sys.argv[1]))
!time -p python tcpu.py 1
!time -p python tcpu.py 2
%%file pcpu.py
from multiprocessing import Pool
import requests
import sys
import time
def saxpy(*args):
s = 0
for i in range(1000):
for j in range(10000):
s += i*j*1.0
return s
def executetask(concurrency, tasksize=10):
argsq = [("http://httpbin.org/html",) for i in range(tasksize)]
pool = Pool(concurrency)
r = pool.map(saxpy, argsq)
print(r)
if __name__ == "__main__":
executetask(int(sys.argv[1]))
!time -p python pcpu.py 1
!time -p python pcpu.py 2
!time -p python pcpu.py 4
from threading import Event
import time
class Clock(threading.Thread):
def setEvent(self, e):
self.stopevent = e
def run(self):
i = 0
while not self.stopevent.isSet():
print("Tick ", i)
i += 1
time.sleep(2)
e = Event()
c = Clock()
c.setEvent(e)
c.start()
e.set()
int("saldk")
%%file functionstest.py
import pytest
def fib(n):
if n in [1,2]:
return 1
return fib(n-1)+ fib(n-2)
def parseint(strnum):
try:
return int(strnum)
except ValueError as v:
if strnum.lower()=="nan" or strnum=="":
return 0
else:
raise v
def test_fib():
assert fib(1) == 1
assert fib(2) == 1
#assert fib(3) == 3
def test_parseint():
assert parseint(" 2") == 2
assert parseint("Nan") == 0
with pytest.raises(ValueError):
parseint("asasd")
!py.test -v functionstest.py
t = datetime.datetime.now()
t.strftime
%%file weekday.py
import datetime
def now():
return datetime.datetime.now()
def weekday():
n = now()
return n.strftime("%A")
import weekday
weekday.weekday()
%%file test_weekday.py
import weekday
import pytest
import datetime
def test_weekday(monkeypatch):
def mydate(y, m, d):
return datetime.datetime(y,m,d)
def fakenow():
return mydate(*date)
monkeypatch.setattr(weekday, "now", fakenow)
date = (2018,4,4)
assert weekday.weekday() == "Wednesday"
date = (2018,2,28)
assert weekday.weekday() == "Wednesday"
!py.test test_weekday.py
help(os.removedirs)
%%file setupteardown.py
import pytest
import os, time
def writefile(filename):
d = [[str(i*j) for i in range(1,5)] for j in range(1,10)]
with open(filename, "w") as f:
for row in d:
f.write(",".join(row))
f.write("\n")
@pytest.fixture
def setup_teardown():
os.mkdir("test")
time.sleep(5)
yield os.path.join(os.getcwd(),"test")
time.sleep(3)
os.removedirs("test")
def test_writefile(setup_teardown):
path = os.path.join(setup_teardown,"data.csv")
writefile(path)
assert os.path.exists(path)
assert os.path.isfile(path)
os.remove(path)
!py.test -v setupteardown.py
import requests
resp = requests.get("http://httpbin.org/html")
resp.status_code
print(resp.text[:100])
resp = requests.get("http://httpbin.org/get", params={"a":"xyz","python":"hello"})
resp.status_code
print(resp.text)
url = "https://duckduckgo.com/"
data = {"q":"pytest","t":"lm","ia":"web"}
resp = requests.get(url, params=data)
resp.status_code
print(resp.text)
resp = requests.post("http://httpbin.org/post", data={"name":"python","passwd":"xxxx"})
print(resp.text)
import requests
url = "https://api.github.com/orgs/vmware/repos"
repos = requests.get(url).json()
type(repos)
repos[0]
for r in repos:
print(r['full_name'], r['forks'], r['watchers'])
for r in sorted(repos, key=lambda r:r['watchers'], reverse=True):
print(r['full_name'], r['forks'], r['watchers'])
Problem
url for map api is "https://maps.googleapis.com/maps/api/distancematrix/json"
parameters of above get url are
origins
destinations
units (metric)