tel: +48 728 438 076
email: piotr.hyzy@eviden.com
Początek¶
Rundka - sprawdzenie mikrofonów¶
Kilka zasad¶
- W razie problemów => chat, potem SMS i telefon, NIE mail
- Materiały szkoleniowe
- Wszystkie pytania sÄ… ok
- Reguła Vegas
- SÅ‚uchawki
- Kamerki
- Chat
- Zgłaszamy wyjścia na początku danego dnia, także pożary, wszystko na chacie
- By default mute podczas wykład
- Przerwy (praca 08:30 - 16:30)
- blok 08:30 - 10:00
- kawowa 10:00 - 10:15 (15')
- blok 10:15 - 11:45
- kawowa 11:45 - 12:00
- blok 12:00 - 13:30
- obiad 13:30 - 14:00
- blok 14:00 - 15:15
- kawowa 15:15 - 15:30
- blok 15:30 - 16:30
- wszystkie czasy sÄ… plus/minus 10'
- Jak zadawać pytanie? 1) przerwanie 2) pytanie na chacie 3) podniesienie wirtualnej ręki
- IDE => dowolne
- Każde ćwiczenie w osobnym pliku/Notebooku
- Nie zapraszamy innych osób
- Zaczynamy punktualnie
- Ćwiczenia w dwójkach, rotacje, ask for help
Powtórka¶
Data Types¶
# int
a = 5
print(a)
print(type(a))
# string
a = '5'
print(a)
print(type(a))
# list
b = [1,2,3,4]
print(b)
print(type(b))
print(b[0])
b[0] = None
print(b[0])
print(b[4])
5 <class 'int'> 5 <class 'str'> [1, 2, 3, 4] <class 'list'> 1 None
--------------------------------------------------------------------------- IndexError Traceback (most recent call last) Cell In[1], line 20 17 b[0] = None 18 print(b[0]) ---> 20 print(b[4]) IndexError: list index out of range
# tuple
c = (1,2,3,4)
print(c)
print(type(c))
print(c[0])
c[0] = None
(1, 2, 3, 4) <class 'tuple'> 1
--------------------------------------------------------------------------- TypeError Traceback (most recent call last) Cell In[2], line 7 4 print(type(c)) 6 print(c[0]) ----> 7 c[0] = None TypeError: 'tuple' object does not support item assignment
# dict
d = {'d': 1, 'e': 2}
print(d)
print(type(d))
{'d': 1, 'e': 2} <class 'dict'>
a_list = [[1,2,3], [4,5,6], [7,8,9], None]
print(len(a_list))
print(a_list[0])
print(a_list[0][1])
4 [1, 2, 3] 2
# tuple of lists and None
b_tuple =([1,2,3], [4,5,6], [7,8,9], None)
for _ele in b_tuple:
print(_ele)
b_tuple[0] = [1,2,3]
b_tuple[0][0] = 2
for _ele in b_tuple:
print(_ele)
[1, 2, 3] [4, 5, 6] [7, 8, 9] None
--------------------------------------------------------------------------- TypeError Traceback (most recent call last) Cell In[5], line 7 4 for _ele in b_tuple: 5 print(_ele) ----> 7 b_tuple[0] = [1,2,3] 9 b_tuple[0][0] = 2 11 for _ele in b_tuple: TypeError: 'tuple' object does not support item assignment
### Function as object
class Foo:
def method1(self):
print('Foo.method1')
def method2(self):
print('Foo.method2')
# class definition
d =[Foo.method1, Foo.method2]
for _ele in d:
print(_ele)
# class instance
foo = Foo()
e =[foo.method1, foo.method2]
for _ele in e:
print(_ele)
_ele()
<function Foo.method1 at 0x10696f250> <function Foo.method2 at 0x10696f520> <bound method Foo.method1 of <__main__.Foo object at 0x106877640>> Foo.method1 <bound method Foo.method2 of <__main__.Foo object at 0x106877640>> Foo.method2
### How call a function
def foo(a, b):
print(a)
print(b)
foo(1, 2) # -> a=1, b=2 # positional arguments
foo(b=1, a=2) # -> a=2, b=1 # named arguments, key-value arguments
1 2 2 1
Functions¶
*args¶
*args in signature¶
# without args
def func(a, b, c):
print(a)
print(b)
print(c)
func(1,2,3)
# func(1, 2)
# func(1,2, 3, 4)
1 2 3
# with args
def func(a, *args):
print(a)
print(args)
print(type(args))
print(f'len: {len(args)}')
func(1,2,3)
1 (2, 3) <class 'tuple'> len: 2
func(1)
func(1, 2)
func(1,2, 3, 4)
1 () <class 'tuple'> len: 0 1 (2,) <class 'tuple'> len: 1 1 (2, 3, 4) <class 'tuple'> len: 3
# args is just keyword, the improtant is *
def foo(*my_attributes):
print(my_attributes)
foo(1,2,3,4,5,6,7,8)
(1, 2, 3, 4, 5, 6, 7, 8)
def boo(a, b, *args):
print(f'{a}, {b}, {args}')
boo (1,2,3)
1, (3,), 2
*args in a call (unpacking)¶
def fun(a, b, c):
print(a, b, c)
list_ = [1, 2, 3]
fun(list_[0], list_[1], list_[2])
fun(*list_)
# *list_ = list_[0], list_[1], list_[2]
1 2 3 1 2 3 my_list [1, 2, 3, 4, 5] 1 [2, 3, 4, 5]
*args in list creation¶
list_ = [3, 4, 5]
my_list = [1, 2, *list_]
print('my_list')
print(my_list)
*args in an assigment¶
list_ = [1, 2, 3, 4, 5]
first, *rest = list_
print(list_)
print(first)
print(rest)
print(type(rest))
### assigment -> unused value
list_ = [1, 2, 3, 4, 5]
first, _, *rest = list_ # best practise to use _ for unused values
print(first)
print(rest)
1 [3, 4, 5]
Exercise: Sum¶
Write a function that sums any number of integer arguments. However, there must be at least one argument. That is, print an error instead of returning zero in the case of no arguments.
### Solution
def my_sum(first, *args):
total = first
for arg in args:
total += arg # => total = total + arg
return total
### Solution 2
def my_sum(first, *args):
return sum([first, *args])
### Expected behaviour
print(my_sum(2, 3, 4)) # ==> 9
print(my_sum(2)) # ==> 2
print(my_sum()) # ==> TypeError
9 2
--------------------------------------------------------------------------- TypeError Traceback (most recent call last) Cell In[16], line 4 2 print(my_sum(2, 3, 4)) # ==> 9 3 print(my_sum(2)) # ==> 2 ----> 4 print(my_sum()) # ==> TypeError TypeError: my_sum() missing 1 required positional argument: 'first'
**kwargs¶
def foo(a,b, **kwargs):
print('a =', a)
print('b =', b)
print('kwargs =', kwargs)
print(type(kwargs))
foo(2, 3, c=4, d=5)
a = 2 b = 3 kwargs = {'c': 4, 'd': 5} <class 'dict'>
Exercise: Keyword Arguments¶
Write function dict_without_Nones that is very similar to dict function, however it ignores keys with None value.
You can do that in at least three different ways:
- using for loop,
- using dictionary comprehension: {: for in if _},
- using filter builtin function.
### Hint -- dict builtin function
a = dict(a=2, b=3, c=None)
print(a)
b = {'a': 2, 'b':3, 'c': None}
print(b)
assert a == b
### Hint -- dict usage
d = {'a': 2, 'b': 3}
print(d)
d['key'] = 'value'
d['c'] = 5
print(d)
{'a': 2, 'b': 3} {'a': 2, 'b': 3, 'key': 'value', 'c': 5}
### Hint -- third way
numbers = [1, 2, 3, 4, 5]
def f(number):
return number%2 == 0
print(list(filter(f, numbers)))
[2, 4]
f = None
g = None
print(id(f), id(g))
# f is g and g is None
r = 1
t = 2
print(id(r), id(t))
4484341808 4484341808 4472013040 4472013072
### Use `a is None` instead of `a == None` because of the following trap:
class A:
def __eq__(self, other):
return True
a = A()
b = None
print(a == b) # => a.__eq__(b)
print(a is b) # id(a) == id(b)
True False
### Solution 1
def dict_without_Nones(**kwargs):
result = kwargs.copy()
for k, v in kwargs.items():
if v is None:
result.pop(k)
return result
### Solution 2
def dict_without_Nones_1(**kwargs):
result = {}
for k, v in kwargs.items():
if v is not None:
result[k] = v
return result
### Solution 3
def dict_without_Nones_2(**kwargs):
return {k: v for k, v in kwargs.items() if v is not None}
### Dict comprehension
# from flat values
print({x: bin(x) for x in range(5)})
# from another dict
input_dict = dict(a=1, b=2, c=3, d=None)
output_dioct = {k: v for k, v in input_dict.items() if v is not None}
### Solution 3
def is_not_None(item):
# item is a pair of (key, value)
key, value = item
return value is not None
def dict_without_Nones_3(**kwargs):
return dict(filter(is_not_None, kwargs.items()))
{0: '0b0', 1: '0b1', 2: '0b10', 3: '0b11', 4: '0b100'}
### Expected behaviour
assert dict_without_Nones(a="1999", b="", c=None) == {'a': '1999', 'b': ''}
%timeit dict_without_Nones(a="1999", b="", c=None)
%timeit dict_without_Nones_1(a="1999", b="", c=None)
%timeit dict_without_Nones_2(a="1999", b="", c=None)
%timeit dict_without_Nones_3(a="1999", b="", c=None)
# ns = 10(-9) nanoseconds
# us = 10(-6) microseconds
972 ns ± 24.8 ns per loop (mean ± std. dev. of 7 runs, 1,000,000 loops each) 918 ns ± 22.2 ns per loop (mean ± std. dev. of 7 runs, 1,000,000 loops each) 1.19 μs ± 5.51 ns per loop (mean ± std. dev. of 7 runs, 1,000,000 loops each) 1.63 μs ± 52.8 ns per loop (mean ± std. dev. of 7 runs, 1,000,000 loops each)
### What timeit does?
from time import time
for run in range(7):
s = time()
for i in range(1000000):
dict_without_Nones(a="1999", b="", c=None)
e = time()
print((e-s)/1000000)
### Using timeit outside IPython and Jupyter Notebook
import timeit
code = 'dict_without_Nones(a="1999", b="", c=None)'
gl = {'dict_without_Nones': dict_without_Nones}
results = timeit.repeat(code, globals=gl)
print(results)
[1.054006638994906, 1.0463737480022246, 1.038764643992181, 1.0419428469904233, 1.0368517390015768]
**kwargs in a call¶
def foo(a, b, c):
print('a =', a)
print('b =', b)
print('c =', c)
d = {'a': 2, 'b': 3, 'c': 4}
foo(**d) # equivalent to foo(a=2, b=3, c=4)
a = 2 b = 3 c = 4
# A trap!
foo(*d)
a = a b = b c = c
Default Values¶
Exercise: Default value trap¶
Implement append_func function that appends an element to a list. The element and the list are arguments of this function. The list should be optional. The default should be an empty list.
### default value
from numpy import cov
def add(a, b, c, convert_to_float=False):
if convert_to_float:
a = float(a)
b = float(b)
c = float(c)
return a + b + c
print(add(1,2,3, convert_to_float=True))
### Wrong solution
def append_func(item, list_=[]):
list_.append(item)
return list_
### Correct solution
def append_func(item, list_=None):
if list_ is None:
list_ = []
list_.append(item)
return list_
6.0
### Expected behaviour
print(append_func('e'))
print(append_func('f'))
['e'] ['f']
assert append_func(3, [1, 2]) == [1, 2, 3]
assert append_func('c', ['a', 'b']) == ['a', 'b', 'c']
assert append_func('e') == ['e']
l = ['a', 'b']
assert append_func('e', l) == ['a', 'b', 'e']
assert l == ['a', 'b', 'e']
assert append_func('f') == ['f']
### This is how Python works
def append_func(item, list_=[]):
list_.append(item)
return list_
# default_list = []
# def append_func(item, list_):
# if list_ nie zostało przekazane:
# list_ = default_list
# list_.append(item)
# return list_
### This is how Python works 2
def generate_list():
print('Generating a new empty list...')
return []
def append_func(item, list_=generate_list()):
list_.append(item)
return list_
### Correct solution
def append_func(item, list_=None):
if list_ is None:
list_ = []
list_.append(item)
return list_
Generating a new empty list...
defaults is a tuple containing default argument values for those arguments that have defaults (or None if no arguments have a default value).
print(append_func.__defaults__)
(None,)
Lambda¶
Essentials¶
items = ['Alice', 'has', 'a', 'cat']
print(sorted(items)) # => A is lower than a as the python string definition says that
print(sorted(items, key=len))
['Alice', 'a', 'cat', 'has'] ['a', 'has', 'cat', 'Alice']
def my_function(item):
return len(item)
print(sorted(items, key=my_function))
# or simpler:
print(sorted(items, key=lambda item: len(item))) # lambda x: func(X)
['a', 'has', 'cat', 'Alice']
Exercise: sort by extension¶
['data', 'txt'] ['text.csv', 'data.csv', 'text.json', 'asd.dfg.json', 'data.txt']
files = ['data.txt', 'text.csv', 'data.csv', 'text.json', 'asd.dfg.json']
# output ['text.csv', 'data.csv', 'text.json', 'asd.dfg.json', 'data.txt']
### Hints
filename = 'data.txt'
print(filename.split('.'))
### Solution
print(sorted(files, key=lambda filename: filename.split('.')[-1]))
Exercise: sort by extension, then by filename¶
('csv', 'data.csv') < ('csv', 'text.csv')
def get_extension(filename):
return filename.split('.')[-1]
print((get_extension('data.txt'), 'data.txt'))
('txt', 'data.txt')
files = ['data.txt', 'text.csv', 'data.csv', 'text.json']
def get_extension(filename):
return filename.split('.')[-1]
print(sorted(files, key=lambda filename: (get_extension(filename), filename)))
['data.csv', 'text.csv', 'text.json', 'data.txt']
Annotation¶
Function Annotation¶
PEP 3107 – Function Annotations
def foo(
a: int,
b: str,
c: dict,
d: list | None = None,
) -> None:
'''
ble ble
'''
pass
foo((1,), (2,), (3,), (4,))
print(foo.__annotations__)
{'a': <class 'int'>, 'b': <class 'str'>, 'c': <class 'dict'>, 'd': list | None, 'return': None}
Variable Annotation¶
PEP 526 – Syntax for Variable Annotations
def foo():
x: int
x = 3
y: float = 2.0
c = 5
def bar():
c: str
print(c)
bar()
--------------------------------------------------------------------------- UnboundLocalError Traceback (most recent call last) Cell In[126], line 12 9 c: str 10 print(c) ---> 12 bar() Cell In[126], line 10, in bar() 8 def bar(): 9 c: str ---> 10 print(c) UnboundLocalError: local variable 'c' referenced before assignment
Decorators¶
Simple Decorator¶
Closures¶
Python closure is a nested function that allows us to access variables of the outer function even after the outer function is closed.
Understanding Closures in Python: A Comprehensive Tutorial
def generate_adder(x):
def add(y):
return x + y
return add
add_5 = generate_adder(5) # add_5 combines add function AND the value of x
print(add_5)
print(add_5(10))
print(add_5(15))
print(add_5.__closure__) # add_5 really is more than just add function
(<cell at 0x11409e2f0: int object at 0x10a8d8170>,)
Exercise: Exploring Nested Functions and Closures in Python¶
In this exercise, you will explore the concept of nested functions and closures in Python. You will create a nested function that performs an operation based on an external parameter and then inspect its closure properties.
tep-by-Step Instructions:
Define the Nested Function: Create a function
outer_function
that takes a single parametern
.Inner Function: Inside
outer_function
, define another functioninner_function
that takes a parameterx
and returns the result of an operation involvingx
andn
.Return the Inner Function: Ensure that
outer_function
returns theinner_function
.Create a Closure: Use
outer_function
to create a closure by passing an argument to it and assigning the result to a variable.Inspect the Closure: Print the
__closure__
attribute of the closure to see the cell contents that are enclosed.Verify the Enclosed Value: Access and print the value stored in the closure's cell contents.
Tasks:
Addition Closure: Create an addition closure where the inner function adds
n
tox
.Subtraction Closure: Create a subtraction closure where the inner function subtracts
n
fromx
.Division Closure: Create a division closure where the inner function divides
x
byn
.Additional Challenge: Modify the inner function to perform a more complex operation, such as raising
x
to the power ofn
, and inspect the closure again.
# Addition closure
def addition(n):
def inner_function(x):
return x + n
return inner_function
add_10 = addition(10)
print(add_10.__closure__)
print(add_10.__closure__[0].cell_contents)
# Subtraction closure
def subtraction(n):
def inner_function(x):
return x - n
return inner_function
subtract_3 = subtraction(3)
print(subtract_3.__closure__)
print(subtract_3.__closure__[0].cell_contents)
# Division closure
def division(n):
def inner_function(x):
return x / n
return inner_function
divide_by_2 = division(2)
print(divide_by_2.__closure__)
print(divide_by_2.__closure__[0].cell_contents)
# Complex operation closure (power)
def power_operation(n):
def inner_function(x):
return x ** n
return inner_function
power_of_4 = power_operation(4)
print(power_of_4.__closure__)
print(power_of_4.__closure__[0].cell_contents)
(<cell at 0x11409e470: int object at 0x10a8d8210>,) 10 (<cell at 0x11409fca0: int object at 0x10a8d8130>,) 3 (<cell at 0x10d7f0250: int object at 0x10a8d8110>,) 2 (<cell at 0x10d7f0220: int object at 0x10a8d8150>,) 4
Decorator Definition¶
import functools
def shouter(func):
print('wywolanie dekorowania funkcji')
@functools.wraps(func) # takes care of func.__doc__ and func.__name__
def wrapper(*args, **kwargs):
print('Before foo')
result = func(*args, **kwargs)
print('After foo')
return result
return wrapper
Decorator Usage¶
@shouter
def foo(a):
'''Docstring'''
print("Message:", a)
return 42
# foo = shouter(foo) # equivalent
result = foo(a="Inside") # delegates to wrapper
print("result =", result)
print(foo.__doc__)
print(foo.__name__)
wywolanie dekorowania funkcji Before foo Message: Inside After foo result = 42 None wrapper
@shouter
def bar(a, b):
print("Bar")
bar(1, 2)
wywolanie dekorowania funkcji Before foo Bar After foo
Exercise: suppress_exceptions¶
Implement suppress_exceptions decorator that catches any exception and converts it to None
### Initial Code
def suppress_exceptions(func):
@functools.wraps(func) # takes care of func.__doc__ and func.__name__
# wrapper definition starts
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception:
return None
# wrapper definiton ends
return wrapper
### Expected Behaviour
@suppress_exceptions
def foo(a, b):
return a/b
print(foo(4, 2)) # ==> 2
print(foo(4, 0)) # ==> return None instead of an exception
2.0 None
Exercise: Using the timeit Decorator¶
In this exercise, you will create a timeit decorator that measures the execution time of a function and displays it after the function completes.
Instructions:
- Implement the timeit decorator, which will measure the execution time of the decorated function and display the execution time in the format Elapsed {time:.3f} sec.
- Use the time function from the time module to measure the time.
- Ensure the decorator correctly handles exceptions and still displays the execution time even if the function fails.
- Apply the decorator to a sample function bar, which introduces delays using sleep and performs simple division.
from time import time, sleep
a = 5
# 1
print('jestem poza foo') # global namespacem kod wykonuje sie od razu (pierwsze przejscie)
### Initial Code
# 2
def suppress_exceptions(func):
# 4, 9
print(f'decor outter: {func.__name__}')
# 5, 10
@functools.wraps(func) # takes care of func.__doc__ and func.__name__
# wrapper definition starts
# 6, 11
def wrapper(*args, **kwargs):
# 13
try:
# 4
print(f'decor inner: {func.__name__}')
return func(*args, **kwargs)
except Exception:
return None
# wrapper definiton ends
# 7
return wrapper
# 3
@suppress_exceptions # ==> foo = supress_exceptions(foo)
def foo():
b = 6
print('jestem w foo') # local namespace funkcji
# 8
@suppress_exceptions # ==> boo = supress_exceptions(boo)
# comment
def boo():
print('jestem w boo')
# 12
foo()
# boo()
jestem poza foo decor outter: foo decor outter: boo decor inner: foo jestem w foo
## Solution
def timeit(func): # func == original bar
@functools.wraps(func) # takes care of func.__doc__ and func.__name__
def wrapper(*args, **kwargs):
start = time() # timestamp
try:
return func(*args, **kwargs)
finally:
elapsed = time() - start # timestamp
print(f"Elapsed {elapsed:.3f} sec")
return wrapper
### Expected behaviour
# @timeit
def bar(a, b):
sleep(0.1)
retval = a / b
sleep(0.1)
return retval
bar = timeit(bar)
# Sample function calls
print(bar(4, 2)) # Expected output: Elapsed 0.204 sec, 2.0
# Elapsed 0.207 sec
# 2.0
print(bar(4, 0)) # Expected output: Elapsed 0.104 sec, Traceback with ZeroDivisionError
# Elapsed 0.104 sec
# raise Zero Division Error
Elapsed 0.207 sec 2.0 Elapsed 0.105 sec
--------------------------------------------------------------------------- ZeroDivisionError Traceback (most recent call last) Cell In[176], line 9 2 print(bar(4, 2)) # Expected output: Elapsed 0.204 sec, 2.0 3 # Elapsed 0.207 sec 4 # 2.0 ----> 9 print(bar(4, 0)) # Expected output: Elapsed 0.104 sec, Traceback with ZeroDivisionError 10 # Elapsed 0.104 sec 11 # raise Zero Division Error Cell In[174], line 7, in timeit.<locals>.wrapper(*args, **kwargs) 5 start = time() # timestamp 6 try: ----> 7 return func(*args, **kwargs) 8 finally: 9 elapsed = time() - start # timestamp Cell In[175], line 5, in bar(a, b) 3 def bar(a, b): 4 sleep(0.1) ----> 5 retval = a / b 6 sleep(0.1) 7 return retval ZeroDivisionError: division by zero
Exercise: Retry Three Times¶
In this exercise, you will create a retry_three_times decorator that will attempt to execute a function up to three times if it raises an exception. If the function fails on the first or second attempt, the decorator will suppress the exception and retry. On the third attempt, it will either return the function's result or let the exception propagate.
Instructions:
- Implement the retry_three_times decorator that retries the execution of the decorated function up to three times.
- Ensure that if the function raises an exception on the first or second attempt, the exception is suppressed and the function is retried.
- If the function raises an exception on the third attempt, the exception should propagate.
- Use functools.wraps(func) to preserve the original function's name and documentation.
### Solution
def retry_three_times(func):
@functools.wraps(func) # takes care of func.__doc__ and func.__name__
def wrapper(*args, **kwargs):
for _ in range(2):
try:
return func(*args, **kwargs)
except Exception:
pass
return func(*args, **kwargs)
return wrapper
### Expected Behaviour 1: Function Succeeds on First Attempt
@retry_three_times
def works():
return 42
print(works())
42
### Expected Behaviour 2: Function Always Fails
@retry_three_times
def always_fails():
raise Exception
print(always_fails())
--------------------------------------------------------------------------- Exception Traceback (most recent call last) Cell In[179], line 6 2 @retry_three_times 3 def always_fails(): 4 raise Exception ----> 6 print(always_fails()) Cell In[177], line 10, in retry_three_times.<locals>.wrapper(*args, **kwargs) 8 except Exception: 9 pass ---> 10 return func(*args, **kwargs) Cell In[179], line 4, in always_fails() 2 @retry_three_times 3 def always_fails(): ----> 4 raise Exception Exception:
### Expected Behaviour 3: Function Succeeds on Third Attempt
trial = 0
@retry_three_times
def works_on_third_trial():
global trial
trial += 1 # trial = trial + 1
if trial == 3:
return 42
else:
raise Exception
print(works_on_third_trial()) # Expected output: 42
42
Tips:
- Use a for loop to handle the retries.
- Make sure the exception is caught and suppressed only on the first and second attempts.
- On the third attempt, either return the function's result or let the exception propagate normally.
Decorators for Registering Functions¶
Exercise (home work): Decorators for Registering Functions¶
In this exercise, you will create a test decorator that registers functions by adding them to a list called test_functions. This will allow you to keep track of which functions have been decorated and then execute them in order.
Instructions:
- Implement a
test
decorator that appends the decorated function to thetest_functions
list. - Ensure that the
test
decorator returns the original function so that it can still be called normally. - Decorate sample functions
a
andb
with thetest
decorator and verify that they are added to thetest_functions
list. - Create a sample function
c
that is not decorated to show the difference. - Assert that the
test_functions
list contains the functionsa
andb
. - Iterate over the
test_functions
list and call each function to demonstrate that they are executed in order.
Iterate over the test_functions list and call each function to demonstrate that they are executed in order.
### Solution
test_functions = []
def test(func):
test_functions.append(func)
return func
import functools
# ### Solution
test_functions = []
def test(func):
test_functions.append(func)
return func
### Expected Behaviour
@test
def a():
print("a()")
@test
def b():
print("b()")
def c():
print("c()")
print('a =', a)
print(test_functions)
assert test_functions == [a, b]
for test_function in test_functions:
test_function()
a = <function a at 0x11305d360> [<function a at 0x11305d360>, <function b at 0x10cdc8c10>] a() b()
Parametrized Decorators¶
Decorator Definition¶
def shouter_ex(msg):
def decor(func): # func == spam
@functools.wraps(func)
def wrapper(*args, **kwargs):
print("Start:", msg)
try:
return func(*args, **kwargs)
finally:
print("End:", msg)
return wrapper
return decor
Decorator Usage¶
@shouter_ex("Decorate spam") # tahe shouter_ex and run it with argument: "Decorate spam" -> return decor => @decor => decor(spam)
def spam(a, b):
'''Docstring'''
return a/b
# is equivalent to: spam = shouter_ex("Decorate spam")(spam)
# or is equivalent to:
dec = shouter_ex("Decorate spam")
# dec == decor
@dec
def spam(a, b):
'''Docstring'''
return a/b
print(spam(4, 2))
Start: Decorate spam End: Decorate spam 2.0
Exercise: Suppress Exceptions 2¶
In this exercise, you will create a decorator suppress_exceptions_ex
that suppresses specified exceptions during the execution of a function. If the specified exception is raised, the decorator should return a predefined value. This allows for graceful handling of errors without interrupting the flow of the program.
Instructions:¶
- Implement the
suppress_exceptions_ex
decorator that takes two optional arguments:exception
: The type of exception to suppress (default is: all exceptions).value_on_exception
: The value to return if the specified exception is raised (default isNone
).
- If no exception is raised, the function should return its normal result.
- Use
functools.wraps
to preserve the original function's metadata.
### SOLUTION
def suppress_exceptions_ex(exception=Exception, value_on_exception=None):
def decor(func): # func == spam
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except exception:
return value_on_exception
return wrapper
return decor
### Expected Behaviour
# BAD # @suppress_exceptions_ex # ==> egg=supress_exception_ex(egg)
# GOOD # @suppress_exceptions_ex() # ==> egg = supress_exceptions_ex()(egg)
@suppress_exceptions_ex(exception=ZeroDivisionError, value_on_exception=0.0)
def egg(a, b):
return a/b
print(egg(4, 2))
print(egg(4, 0)) # ==> return 0 instead ZeroDivision Error
print(egg('asdf', 'qwer')) # ==> TypeError
2.0 None None
Object Oriented Programming¶
Class & Object Attributes¶
Example¶
class Person: # class header
phone = None # class atrribute (variable)
phones = [] # class atrribute (variable)
# __init__ is a one of first functions called during class instantiation
def __init__(self, name): # method definition
self.name = name # instance attribute
p1 = Person('John')
p2 = Person('Alice')
print(p1.name)
print(p2.name)
p1.name = 'Bob'
print(p1.name)
print(p2.name)
John Alice Bob Alice
### Case 1
class Person:
phone = None
def __init__(self, name):
self.name = name
def mymethod(self):
pass
p1 = Person('John')
p2 = Person('Alice')
print('p1.phone =', p1.phone)
p1.phone = 98765
print('p1.phone =', p1.phone)
print('p2.phone =', p2.phone)
p1.phone = None p1.phone = 98765 p2.phone = None
print('p1.__dict__ =', p1.__dict__)
print('p2.__dict__ =', p2.__dict__)
print('Person.__dict__ =', Person.__dict__)
p2.mymethod()
p1.__dict__ = {'name': 'John', 'phone': 98765} p2.__dict__ = {'name': 'Alice'} Person.__dict__ = {'__module__': '__main__', 'phone': None, '__init__': <function Person.__init__ at 0x114d755a0>, 'mymethod': <function Person.mymethod at 0x114d74790>, '__dict__': <attribute '__dict__' of 'Person' objects>, '__weakref__': <attribute '__weakref__' of 'Person' objects>, '__doc__': None}
### Case 2
class Person:
phones = []
instances = []
def __init__(self, name):
self.name = name
p1 = Person('John')
p2 = Person('Alice')
print('p1.phones =', p1.phones)
p1.phones = []
p1.phones.append(987654)
print('p1.phones =', p1.phones)
print('p2.phones =', p2.phones)
p1.phones = [987654] p2.phones = [987654]
print('p1.__dict__ =', p1.__dict__)
print('p2.__dict__ =', p2.__dict__)
print('Person.__dict__ =', Person.__dict__)
p1.__dict__ = {'name': 'John'} p2.__dict__ = {'name': 'Alice'} Person.__dict__ = {'__module__': '__main__', 'phones': [987654], '__init__': <function Person.__init__ at 0x114d74670>, '__dict__': <attribute '__dict__' of 'Person' objects>, '__weakref__': <attribute '__weakref__' of 'Person' objects>, '__doc__': None}
### Case 3
class Person:
phones = []
def __init__(self, name):
self.name = name
p1 = Person('John')
p2 = Person('Alice')
print('p1.phones =', p1.phones)
p1.phones = []
p1.phones = [98765]
p1.phones.append(12345)
print('p1.phones =', p1.phones)
print('p2.phones =', p2.phones)
p1.phones = [98765, 12345] p2.phones = []
print('p1.__dict__ =', p1.__dict__)
print('p2.__dict__ =', p2.__dict__)
print('Person.__dict__ =', Person.__dict__)
p1.__dict__ = {'name': 'John', 'phones': [98765]} p2.__dict__ = {'name': 'Alice'} Person.__dict__ = {'__module__': '__main__', 'phones': [], '__init__': <function Person.__init__ at 0x114d75480>, '__dict__': <attribute '__dict__' of 'Person' objects>, '__weakref__': <attribute '__weakref__' of 'Person' objects>, '__doc__': None}
print(Person.phones)
print(p1.__class__.phones) # informational
print(p1.phones)
[] [] [98765, 12345]
Access Control¶
Protected Attributes¶
class BankAccount:
def __init__(self, initial_balance):
self._balance = initial_balance
def print_balance(self):
print(self._balance)
def get_balance(self):
return self._balance
account = BankAccount(100.0)
print(account.get_balance()) # proper usage
print(account._balance) # really no protection, it's only a convention
print(account.__dict__)
100.0 100.0 {'_balance': 100.0}
Private Attributes¶
class BankAccount:
def __init__(self, initial_balance=0):
self.__balance = initial_balance
def withdraw(self, amount):
self.__balance -= amount
def deposit(self, amount):
self.__balance += amount
def info(self):
print(f"Balance is {self.__balance}")
account = BankAccount(100.0)
print(account.__dict__)
# print(account.__balance) # => AttributeError
# print(account._BankAccount__balance)
{'_BankAccount__balance': 100.0} 100.0
@property¶
Read-only Property¶
class Point:
def __init__(self, x, y):
self.x = x
self.y = y
@property
def is_zero(self):
return self.x == 0 and self.y == 0
p1 = Point(2, 3)
p2 = Point(0, 0)
print(p1.is_zero)
print(p2.is_zero)
# p1.is_zero = False # ==> Atribute Error - read only
False True
--------------------------------------------------------------------------- AttributeError Traceback (most recent call last) Cell In[253], line 7 4 print(p1.is_zero) 5 print(p2.is_zero) ----> 7 p1.is_zero = False AttributeError: can't set attribute 'is_zero'
Read and Write Property¶
from math import sqrt
class Vector:
def __init__(self, x, y): # always starts in 0,0
self.x = x
self.y = y
# getter (read-only property)
@property # telloing this is getter (read-only)
def length(self): # method definiton with a name of our dynamic attribute - property
return sqrt(self.x**2 + self.y**2) # returned value, if I wish to keep it as internal variable, the best practise is to use _name
# setter (requires getter)
@length.setter # must be the same name (length) as above
def length(self, new_length): # the same here - must be the same name (length) as above
if new_length < 0:
raise ValueError("Negative length is invalid")
elif new_length == 0:
self.x = 0
self.y = 0
elif self.length == 0:
raise ValueError("Ooops, unknown direction")
else:
factor = new_length / self.length
self.x *= factor
self.y *= factor
v = Vector(3, 4)
print(v.length)
5.0
v.length = 10 # delegates to Vector.length(self=v, new_length=10)
print(v.length, v.x, v.y)
10.0 6.0 8.0
### Expected Behaviour
v = Vector(3, 4)
v.length = 10
assert (v.x, v.y) == (6, 8)
u = Vector(0, 0)
u.length = 0
assert (u.x, u.y) == (0, 0)
# u.length = 10 # ==> ValueError
Exercise: Refactoring¶¶
### Before
class Connection:
def __init__(self, target):
self.target = target
self.state = 'disconnected'
def connect(self):
ip, port = self.target.split("::")
print(f"Connecting to ip={ip} port={port}")
self.state = 'connected'
def disconnect(self):
ip, port = self.target.split("::")
print(f"Disconnecting from ip={ip} port={port}")
self.state = 'disconnected'
c = Connection('127.0.0.1::80')
c.target = '127.0.0.1::60' # overwrite target - should be allowed
assert c.target == '127.0.0.1::60'
assert c.state == 'disconnected'
c.connect()
assert c.state == 'connected'
c.target = '127.0.0.1::40' # should be disallowed
c.disconnect()
c.state = 'connected' # should be disallowed
# target writable in dosconnected tsate only
# state is private (read-only) to class
Connecting to ip=127.0.0.1 port=60 Disconnecting from ip=127.0.0.1 port=40
### After
class Connection:
def __init__(self, target):
self._connected = False
self.target = target # delegates to target setter
def connect(self):
print(f"Connecting to ip={self._ip} port={self._port}")
self._connected = True
def disconnect(self):
print(f"Disconnecting from ip={self._ip} port={self._port}")
self._connected = False
@property
def target(self):
return f"{self._ip}::{self._port}"
@target.setter
def target(self, new_target):
if self._connected:
raise ValueError("Cannot change target when connected")
self._ip, self._port = new_target.split("::")
@property
def state(self):
# if self._connected:
# return 'connected'
# else:
# return 'disconnected'
return 'connected' if self._connected else 'disconnected'
### Expected Behaviour
c = Connection('127.0.0.1::80')
c.target = '127.0.0.1::60'
assert c.target == '127.0.0.1::60'
assert c.state == 'disconnected'
c.connect()
assert c.state == 'connected'
# c.target = '127.0.0.1::40' # ==> ValueError
c.disconnect()
# c.state = 'connected' # ==> AttributeError
assert set(c.__dict__.keys()) == {'_ip', '_port', '_connected'}
# c.target and c.state should be properties
# c.target should be writtable only when disconnected
# c.state should be read-only
Connecting to ip=127.0.0.1 port=60 Disconnecting from ip=127.0.0.1 port=60
Class as a Function¶
Basics¶
class Factorial:
def __init__(self):
self.cache = {0: 1}
def __call__(self, n): # it can take any arguments, similar to a regullar function
if n not in self.cache:
print(f'Computing fact({n})')
self.cache[n] = n * self(n-1)
return self.cache[n]
fact = Factorial() # delegates to __init__ and creates new object
print(fact(3)) # delegates fact.__call__(3)
print(fact(3)) # delegates fact.__call__(3)
Computing fact(3) Computing fact(2) Computing fact(1) 6 6
print(fact.cache)
{0: 1, 1: 1, 2: 2, 3: 6}
assert fact(3) == 3 * 2 * 1
assert fact(5) == 5 * fact(4)
Computing fact(5) Computing fact(4)
print(fact.cache)
{0: 1, 1: 1, 2: 2, 3: 6, 4: 24, 5: 120}
Exercise: CachedFunction class¶¶
Implement a class that let you cache any function.
# Cache of caches
class CachedFunction:
class _SingleFunction:
def __init__(self, function):
self._function = function
self._cache = {}
def __call__(self, *args):
if args not in self._cache:
self._cache[args] = self._function(*args)
return self._cache[args]
def __init__(self):
self.functions = {}
def __call__(self, *args):
## ?? which function is triggered
if args not in self._cache:
self._cache[args] = self._function(*args)
return self._cache[args]
def add_function(self, func):
if func not in self.functions:
self.functions[func] = self._SingleFunction(func)
return
# SOLUtiON !!!
# Your Code Here
class CachedFunction:
def __init__(self, function):
self._function = function
self._cache = {}
def __call__(self, *args):
# ask for permission
# if args not in self._cache:
# self._cache[args] = self._function(*args)
# return self._cache[args]
# ask for forgiveness
try:
return self._cache[args]
except KeyError:
ret = self._cache[args] = self._function(*args)
return ret
except TypeError: # when args unhashable
return self._function(*args)
### Expected Behaviour
from time import sleep
def long_computations(a, b):
print("Computations going on...")
sleep(1)
return a + b
cached_long_computation = CachedFunction(long_computations) # calls __init__
def foo():
pass
cached_foo = CachedFunction(foo)
# Imagine there are more functions we need to cache,
# so we don't want to put the caching logic inside
# long_computations function.
print(cached_long_computation(2, 3)) # calls __call__(2, 3) takes 10s
Computations going on... 5
print(cached_long_computation(2, 3)) # no computations
5
@CachedFunction
def cached_inc(a):
print(f'Executing inc({a})')
sleep(2)
return a + 1
cached_inc = CachedFunction(cached_inc)
print(cached_inc(3))
print(cached_inc(3))
Executing inc(3) 4 4
Ask for permission vs forgiveness¶
Ask for permission¶
if plik istnieje:
# inny proces usuwa plik
otworz plik # => błąd
Ask for forgiveness¶
try: s = open('file.txt') except FileNotFoundError: print("asdf")
### Ask for permission
numbers = [42, 20, 0]
if len(numbers) > 2:
# 100 linii kodu
third = numbers[2]
else:
third = None
# Ask for forgiveness
try:
third = numbers[2]
inversion = 1 / third
except IndexError:
third = None
inversion = None
except ZeroDivisionError:
print("zero division")
Single Inheritance¶
Example¶¶
class Base: # inherits implicitly from object = Base(object):
def base_method(self):
print('base_method')
class Child(Base):
def child_method(self):
print('child_method')
child = Child()
child.child_method()
child.base_method()
child_method base_method
# Where do we look for base_method?
print(child.__dict__)
print(Child.__dict__)
print(Base.__dict__)
{} {'__module__': '__main__', 'child_method': <function Child.child_method at 0x117589240>, '__doc__': None} {'__module__': '__main__', 'base_method': <function Base.base_method at 0x117589ea0>, '__dict__': <attribute '__dict__' of 'Base' objects>, '__weakref__': <attribute '__weakref__' of 'Base' objects>, '__doc__': None}
base = Base()
base.base_method()
base.child_method()
base_method
--------------------------------------------------------------------------- AttributeError Traceback (most recent call last) Cell In[301], line 3 1 base = Base() 2 base.base_method() ----> 3 base.child_method() AttributeError: 'Base' object has no attribute 'child_method'
Overidden and super()¶
class Base:
def foo(self):
print('Base.foo')
def bar(self):
print('Base.bar')
def base_method(self):
print('Base.base_method')
def foo2(self):
raise NotImplementedError # tell other developers that they have to implement the method when inheriting the class
class Child(Base):
def foo(self):
super().foo()
print('Child.foo')
def bar(self):
print('Child.bar')
child = Child()
child.bar()
Child.bar
child.foo()
Base.foo Child.foo
child.base_method()
Base.base_method
child.foo2()
--------------------------------------------------------------------------- NotImplementedError Traceback (most recent call last) Cell In[309], line 1 ----> 1 child.foo2() Cell In[305], line 12, in Base.foo2(self) 11 def foo2(self): ---> 12 raise NotImplementedError NotImplementedError:
class Point:
a = 0
def __init__(self, x, y):
self.x = x
self.y = y
class Point3D(Point):
def __init__(self, x, y, z):
super().__init__(x, y) # run overoaded function from parent on child instance
self.z = z
p3D = Point3D(2, 3, 4)
print(p3D.__dict__)
print(p3D.z)
print(p3D.a)
{'x': 2, 'y': 3, 'z': 4} 4 0
Inheriting from object¶
The dir() method returns the list of valid attributes of the passed object.
class MyObject: # implicitly inherits from object
pass
mo = MyObject()
print(dir(mo)) # all attributes of an object
['__class__', '__delattr__', '__dict__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__le__', '__lt__', '__module__', '__ne__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', '__weakref__']
print(mo.__str__())
print(str(mo))
print(mo)
<__main__.MyObject object at 0x114c40af0> <__main__.MyObject object at 0x114c40af0> <__main__.MyObject object at 0x114c40af0>
### Initial Code
class MyObject:
def __str__(self):
default = super().__str__()
return (f"The default __str__ result is {default!r}, "
"however here we modified it.")
o = object()
print(o)
mo = MyObject()
print(mo)
<object object at 0x113a8ac80> The default __str__ result is '<__main__.MyObject object at 0x114c414e0>', however here we modified it.
mo = MyObject()
MyObject.__str__ = lambda self: "qwer"
print(mo)
qwer
x = MyObject()
print(str(x)) # x.__str__()
print(repr(x)) # x.__repr__()
qwer <__main__.MyObject object at 0x114c401c0>
When Should You Use .repr() vs .str() in Python?
DefaultDict Use Cases¶
### 1st Use Case - Counting Distinct Objects
fruits = ['apple', 'orange', 'apple']
counts = {}
for key in fruits:
if key not in counts:
counts[key] = 0
counts[key] += 1 # counts[key] = counts[key] + 1
# or counts[key] = counts.get(key, 0) + 1
assert counts == {'apple': 2, 'orange': 1}
from collections import defaultdict
fruits = ['apple', 'orange', 'apple']
counts = defaultdict(int) # int() == 0
# counts = defaultdict(lambda: 0)
for key in fruits:
counts[key] += 1
assert counts == {'apple': 2, 'orange': 1}
### 2nd Use Case - MultiDictionary:
fruits = ['apple', 'orange', 'banana']
d = {}
for value in fruits:
key = len(value)
if key not in d:
d[key] = []
d[key].append(value)
assert d == {5: ['apple'], 6: ['orange', 'banana']}
fruits = ['apple', 'orange', 'banana']
d = defaultdict(list)
# d = defaultdict(lambda: [])
for value in fruits:
key = len(value)
d[key].append(value)
assert d == {5: ['apple'], 6: ['orange', 'banana']}
print(d)
defaultdict(<class 'list'>, {5: ['apple'], 6: ['orange', 'banana']})
Exercise: DefaultDict Implementation¶
Implement DefaultDict that mimics defaultdict. Use inheritance instead of composition.
class Foo: def getitem(self, key): ... return value
foo = Foo() foo['key']==> foo.getitem('key')
class Foo:
def __getitem__(self, key):
return None
d = Foo()
print(d[1])
None
### Initial Code
class DefaultDict(dict):
def __init__(self, init_value_factory):
self._init_value_factory = init_value_factory
def __getitem__(self, key): # called by d[key] syntax
### Ask for permission
# if key not in self:
# self[key] = self._init_value_factory()
# return super().__getitem__(key)
# # return super()[key] # doesn't work
### Ask for forgiveness
try:
return super().__getitem__(key)
except KeyError:
value = self[key] = self._init_value_factory()
return value
### Solution 3
# __missing__ (key)
# This method is called by the __getitem__() method of the dict class
# when the requested key is not found; whatever it returns or raises is
# then returned or raised by __getitem__() . Note that __missing__() is not
# called for any operations besides __getitem__()
class DefaultDict(dict):
def __init__(self, init_value_factory):
self._init_value_factory = init_value_factory
def __missing__(self, key):
value = self[key] = self._init_value_factory()
return value
### Expected Behaviour
d = DefaultDict(list)
assert d['a'] == []
d['b'].append(2)
assert d['b'] == [2]
d = DefaultDict(int)
d['asdf'] += 1
assert d['asdf'] == 1
Special Methods¶
Examples¶
class Foo:
def __init__(self):
print("Constructor")
def __str__(self):
return "Foo!!!"
def __getitem__(self, key):
if key >= 5:
raise IndexError
return key + 100
# def __len__(self):
# return 5
f = Foo() # f = Foo.__init__()
print(str(f)) # f.__str__()
print(f[3]) # f.__getitem__(3)
# print(len(f))
Constructor Foo!!! 103
for item in f: # gets item[0], item[1] and so on until an error occures
print(item)
100 101 102 103 104
print(list(f))
[100, 101, 102, 103, 104]
Exercise: Account with Transaction History¶
# SOLUTION
class Account:
def __init__(self, initial_balance=0):
self._balance = initial_balance
self._transactions = []
def deposit(self, amount):
self._balance += amount
self._transactions.append(amount)
def withdraw(self, amount):
self._balance -= amount
self._transactions.append(-amount)
def __getitem__(self, index):
return self._transactions[index]
@property
def balance(self):
return self._balance
### Expected behaviour
acc = Account(initial_balance=0)
assert acc.balance == 0
# acc.balance = 0 # ==> AttributeError: can't set a property
acc.deposit(1000)
acc.withdraw(500)
acc.deposit(1000)
assert acc.balance == 1500
assert acc[0] == 1000
assert acc[1] == -500
assert acc[2] == 1000
# acc[3] # ==> IndexError
for transaction in acc:
print(transaction)
print("Transactions:", list(acc))
1000 -500 1000 Transactions: [1000, -500, 1000]
1000 -500 1000 Transactions: [1000, -500, 1000]
class Boo:
def __init__(self, value):
self.value = value
def __eq__(self, other):
return self.value == other.value
def __lt__(self, other):
return self.value < other.value
def __str__(self):
return self.value
def __repr__(self):
return str(self.value)
b1 = Boo(6)
b2 = Boo(5)
print(b1 == b2)
print(sorted([b1,b2]))
print(b1 > b2)
False [5, 6] True
Exercise: Comparable Accounts¶
@functools.total_ordering Given a class defining one or more rich comparison ordering methods, this class decorator supplies the rest. This simplifies the effort involved in specifying all of the possible rich comparison operations:
The class must define one of lt(), le(), gt(), or ge(). In addition, the class should supply an eq() method.
### Solution
from functools import total_ordering
@total_ordering
class Account:
def __init__(self, initial_balance=0):
self._balance = initial_balance
def __eq__(self, other):
return self._balance == other._balance
def __lt__(self, other):
return self._balance < other._balance
### Expected Behaviour
acc1 = Account(initial_balance=1000)
acc2 = Account(initial_balance=1000)
print(acc1 == acc2) # acc1.__eq__(acc2)
print(acc1 != acc2) # acc1.__ne__(acc2) # not acc1 == acc2
print(acc1 > acc2) # acc1.__gt__(acc2) # not (acc1 < acc2 or acc1 == acc2)
print(acc1 >= acc2) # acc1.__ge__(acc2) # not acc1 < acc2
print(acc1 < acc2) # acc1.__lt__(acc2)
print(acc1 <= acc2) # acc1.__le__(acc2) # acc1 < acc2 or acc1 == acc2
True False False True False True
### Expected Behaviour
acc1 = Account(initial_balance=1000)
acc2 = Account(initial_balance=1500)
assert acc1 < acc2 # delegates to acc1.__lt__(acc2)
assert acc2 > acc1
dataclasses¶
Example¶
# pip install dataclasses # for Python 3.6 and earlier
from dataclasses import dataclass # backport for Python 3.6 and earlier
@dataclass
class Point:
x: float
y: float
z: float = 0
p = Point(2, 3, 0)
print(p.x)
print(p.y)
print(p.z)
print(p)
2 3 0 Point(x=2, y=3, z=0)
r = Point(2, 3) # Uses default value z=0
print(r)
Point(x=2, y=3, z=0)
print(p == r) # Automagical support for comparisions
True
Exercise (home): Account with Enhanced Transaction History¶
### Initial Code
from dataclasses import dataclass
from datetime import datetime
# Hint: datetime.now() to get current time
print(repr(datetime.now()))
datetime.datetime(2024, 6, 24, 9, 29, 21, 709109)
# SOLUTION
from dataclasses import field
@dataclass
class Deposit:
amount: float
# time: datetime = field(default_factory=lambda: datetime.now())
time: datetime = field(default_factory=datetime.now) # ==> pass definiton of datetime.now
@dataclass
class Withdraw:
amount: float
time: datetime = None # field(default_factory=datetime.now)
def __post_init__(self):
if self.time is None:
self.time = datetime.now()
class Account:
def __init__(self, initial_balance=0):
self._balance = initial_balance
self._transactions = []
def deposit(self, amount):
self._balance += amount
transaction = Deposit(amount)
self._transactions.append(transaction)
def withdraw(self, amount):
self._balance -= amount
transaction = Withdraw(amount)
self._transactions.append(transaction)
def __getitem__(self, index):
return self._transactions[index]
@property
def balance(self):
return self._balance
### Expected behaviour
acc = Account(initial_balance=0)
assert acc.balance == 0
# acc.balance = 0 # ==> AttributeError: can't set a property
acc.deposit(1000)
acc.withdraw(500)
acc.deposit(1000)
assert acc.balance == 1500
for transaction in acc:
print(transaction)
Deposit(amount=1000, time=datetime.datetime(2024, 6, 24, 9, 33, 6, 8454)) Withdraw(amount=500, time=datetime.datetime(2024, 6, 24, 9, 33, 6, 8497)) Deposit(amount=1000, time=datetime.datetime(2024, 6, 24, 9, 33, 6, 8596))
Deposit(amount=1000, time=datetime.datetime(2024, 6, 17, 19, 59, 28, 4734))
Withdraw(amount=500, time=datetime.datetime(2024, 6, 17, 19, 59, 28, 4773))
Deposit(amount=1000, time=datetime.datetime(2024, 6, 17, 19, 59, 28, 4804))
@classmethod¶
@classmethod vs Method¶
class Foo:
numbers = []
def method(self, a): # self = instance of Foo created by instance = Foo()
print(self, a)
@classmethod
def class_method(cls, a): # cls == Foo
print(cls, a)
foo = Foo()
foo.method(2)
Foo.class_method((2))
<__main__.Foo object at 0x117665a20> 2 <class '__main__.Foo'> 2
foo.method(42) # equivalent to Foo.method(f, 42)
foo.class_method(42) # useless most of the time
Foo.method(f, 42)
Foo.class_method(42)
<__main__.Foo object at 0x117665a20> 42 <class '__main__.Foo'> 42 Foo!!! 42 <class '__main__.Foo'> 42
from datetime import datetime
print(datetime.now())
print(datetime(2020, 1, 22))
2024-06-18 16:24:37.853944 2020-01-22 00:00:00
Exercise: Date.from_string (home work)¶
@dataclass
class Date:
day: int
month: int
year: int
@classmethod
def from_string(cls, date_as_string):
day, month, year = date_as_string.split('-')
return cls(int(day), int(month), int(year)) # Date.__init__(malloc(), day, month, year)
### Expected Behaviour
d1 = Date(20, 1, 2016) # one way to create a Date object
d2 = Date.from_string('20-01-2016') # another way
assert isinstance(d1.day, int)
assert (d1.day, d1.month, d1.year) == (20, 1, 2016)
assert (d2.day, d2.month, d2.year) == (20, 1, 2016)
REST APIs & Web Programming¶
Working with JSON¶
JSON Data Types¶
[from Wikipedia] JSON's basic data types are:
Number: a signed decimal number that may contain a fractional part and may use exponential E notation, but cannot include non-numbers such as NaN. The format makes no distinction between integer and floating-point. JavaScript uses a double-precision floating-point format for all its numeric values, but other languages implementing JSON may encode numbers differently.
String: a sequence of zero or more Unicode characters. Strings are delimited with double-quotation marks and support a backslash escaping syntax.
Boolean: either of the values true or false
Array: an ordered list of zero or more values, each of which may be of any type. Arrays use square bracket notation with comma-separated elements.
Object: an unordered collection of name–value pairs where the names (also called keys) are strings. Objects are intended to represent associative arrays, where each key is unique within an object. Objects are delimited with curly brackets and use commas to separate each pair, while within each pair the colon ':' character separates the key or name from its value.
null: an empty value, using the word null
Whitespace is allowed and ignored around or between syntactic elements (values and punctuation, but not within a string value). Four specific characters are considered whitespace for this purpose: space, horizontal tab, line feed, and carriage return. In particular, the byte order mark must not be generated by a conforming implementation (though it may be accepted when parsing JSON). JSON does not provide syntax for comments.
Early versions of JSON (such as specified by RFC 4627) required that a valid JSON text must consist of only an object or an array type, which could contain other types within them.
Example JSON¶
{ "firstName": "John", "lastName": "Smith", "isAlive": true, "age": 27, "address": { "streetAddress": "21 2nd Street", "city": "New York", "state": "NY", "postalCode": "10021-3100" }, "phoneNumbers": [ { "type": "home", "number": "212 555-1234" }, { "type": "office", "number": "646 555-4567" }, { "type": "mobile", "number": "123 456-7890" } ], "children": [], "spouse": null }
{
"firstName": "John",
"lastName": "Smith",
"isAlive": true,
"age": 27,
"address": {
"streetAddress": "21 2nd Street",
"city": "New York",
"state": "NY",
"postalCode": "10021-3100"
},
"phoneNumbers": [
{
"type": "home",
"number": "212 555-1234"
},
{
"type": "office",
"number": "646 555-4567"
},
{
"type": "mobile",
"number": "123 456-7890"
}
],
"children": [],
"spouse": null
}
Processing Complex JSON¶
%%writefile data.json
{
"hasMore": false,
"participantList": [
{
"displayName": "Piotr Hyzy",
"firstName": "Piotr",
"isDeleted": false,
"lastName": "Hyzy",
"type": "REGULAR",
"userId": "35595c57-8871-44d8-9bf3-031f876c765e"
},
{
"displayName": "Lukasz Kowalski",
"firstName": "Lukasz",
"isDeleted": false,
"lastName": "Kowalskit",
"type": "REGULAR",
"userId": "5aa65e6e-e450-4c95-839d-c8a973d18a51"
}
]
}
Overwriting data.json
import json
with open('data.json', 'r') as file:
# data = json.load(file) # file as string
data = json.loads(file.read()) # file as stream
print(data)
print(type(data))
{'hasMore': False, 'participantList': [{'displayName': 'Piotr Hyzy', 'firstName': 'Piotr', 'isDeleted': False, 'lastName': 'Hyzy', 'type': 'REGULAR', 'userId': '35595c57-8871-44d8-9bf3-031f876c765e'}, {'displayName': 'Lukasz Kowalski', 'firstName': 'Lukasz', 'isDeleted': False, 'lastName': 'Kowalskit', 'type': 'REGULAR', 'userId': '5aa65e6e-e450-4c95-839d-c8a973d18a51'}]} <class 'dict'>
user_by_id = {
record["userId"]: record
for record in data["participantList"]
}
for id_, user in user_by_id.items():
print(f"{id_} -> {user}")
35595c57-8871-44d8-9bf3-031f876c765e -> {'displayName': 'Piotr Hyzy', 'firstName': 'Piotr', 'isDeleted': False, 'lastName': 'Hyzy', 'type': 'REGULAR', 'userId': '35595c57-8871-44d8-9bf3-031f876c765e'} 5aa65e6e-e450-4c95-839d-c8a973d18a51 -> {'displayName': 'Lukasz Kowalski', 'firstName': 'Lukasz', 'isDeleted': False, 'lastName': 'Kowalskit', 'type': 'REGULAR', 'userId': '5aa65e6e-e450-4c95-839d-c8a973d18a51'}
user_id = "35595c57-8871-44d8-9bf3-031f876c765e"
print(user_by_id[user_id]["firstName"]) # print firstName of user_id
Piotr
Object Indexing¶
Object indexing is a programming technique used to improve the efficiency of object lookup and retrieval. This technique involves creating a data structure, typically a dictionary in Python, where the keys are attributes of the objects, and the values are the objects themselves. This allows for fast and efficient access to objects based on their attribute values.
In Python, dictionaries are well-suited for this purpose due to their ability to provide average-case O(1) time complexity for lookups. By mapping an object's attribute to the object itself, you can quickly retrieve the object without having to iterate through a list.
# We have a list of objects
class Person:
def __init__(self, id, name):
self.id = id
self.name = name
# Creating a list of Person objects
people = [
Person(1, 'Alice'),
Person(2, 'Bob'),
Person(3, 'Charlie'),
]
# Creating a dictionary where the key is the 'id' attribute and the value is the entire object
people_dict = {person.id: person for person in people}
# Now we can quickly look up an object by its 'id'
person = people_dict[3]
print(person.name) # Output: Bob
Charlie
In this example, people_dict is a dictionary where the keys are id values and the values are the corresponding Person objects. This allows us to quickly find a Person object by its id attribute.
- Benefits of Object Indexing
- Efficiency: Object indexing significantly reduces the time complexity of lookups from O(n) to O(1), where n is the number of objects.
- Simplicity: Using dictionaries or similar data structures makes the code easier to understand and maintain.
- Scalability: As the number of objects grows, the performance benefit of using indexing becomes more pronounced.
Exercise: Interfaces¶
https://hyzy.dev/trainings/interface-data.json
Generate the following raport:
DN | Description | Speed | MTU |
---|---|---|---|
topology/pod-1/node-201/sys/phys-[eth1/33] | inherit | 9150 | |
topology/pod-1/node-201/sys/phys-[eth1/34] | inherit | 9150 | |
topology/pod-1/node-201/sys/phys-[eth1/35] | inherit | 9150 |
! wget https://hyzy.dev/trainings/interface-data.json
### Solution
import json
with open('interface-data.json') as stream:
data = json.load(stream)
print("DN Description Speed MTU")
print("-------------------------------------------------- -------------------- ------- ------")
for interface in data["imdata"][:10]: # loop over data.imdata for first 10 elements
# print(interface)
_interface = interface["l1PhysIf"]["attributes"]
# print(_interface)
dn = _interface["dn"]
descr = _interface["descr"]
speed = _interface["speed"]
mtu = _interface["mtu"]
# print fields formatted in columns
print(f"{dn:50} {descr:20} {speed:7} {mtu:6}")
DN Description Speed MTU -------------------------------------------------- -------------------- ------- ------ topology/pod-1/node-201/sys/phys-[eth1/33] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/34] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/35] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/36] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/1] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/2] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/3] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/4] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/5] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/6] inherit 9150
### Solution 2
import json
with open('interface-data.json') as stream:
data = json.load(stream)
interfaces = [record["l1PhysIf"]["attributes"] for record in data["imdata"]]
# dns = [interface['dn'] for interface in interfaces]
print("DN Description Speed MTU")
print("-------------------------------------------------- -------------------- ------- ------")
for interface in interfaces[:10]:
#print(interface)
dn = interface["dn"]
descr = interface["descr"]
speed = interface["speed"]
mtu = interface["mtu"]
# print fields formatted in columns
print(f"{dn:50} {descr:20} {speed:7} {mtu:6}")
DN Description Speed MTU -------------------------------------------------- -------------------- ------- ------ topology/pod-1/node-201/sys/phys-[eth1/33] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/34] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/35] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/36] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/1] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/2] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/3] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/4] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/5] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/6] inherit 9150
### Solution 3
import json
with open('interface-data.json') as stream:
data = json.load(stream)
def print_int(dn, descr, speed, mtu):
print(f"{dn:50} {descr:20} {speed:7} {mtu:6}")
# implit
# return None
def translate_int(interface):
return interface["dn"], interface["descr"], interface["speed"], interface["mtu"]
print("DN Description Speed MTU")
print("-------------------------------------------------- -------------------- ------- ------")
_= [print_int(*translate_int(record["l1PhysIf"]["attributes"])) for record in data["imdata"]]
# print(_)
DN Description Speed MTU -------------------------------------------------- -------------------- ------- ------ topology/pod-1/node-201/sys/phys-[eth1/33] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/34] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/35] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/36] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/1] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/2] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/3] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/4] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/5] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/6] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/7] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/8] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/9] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/10] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/11] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/12] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/13] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/14] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/15] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/16] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/17] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/18] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/19] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/20] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/21] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/22] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/23] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/24] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/25] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/26] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/27] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/28] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/29] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/30] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/31] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/32] inherit 9150 topology/pod-1/node-103/sys/phys-[eth1/97] inherit 9150 topology/pod-1/node-103/sys/phys-[eth1/98] inherit 9150 topology/pod-1/node-103/sys/phys-[eth1/99] inherit 9150 topology/pod-1/node-103/sys/phys-[eth1/100] inherit 9150 topology/pod-1/node-103/sys/phys-[eth1/101] inherit 9150 topology/pod-1/node-103/sys/phys-[eth1/102] inherit 9150 topology/pod-1/node-103/sys/phys-[eth1/103] inherit 9150 topology/pod-1/node-103/sys/phys-[eth1/104] inherit 9150 topology/pod-1/node-103/sys/phys-[eth1/1] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/2] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/3] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/4] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/5] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/6] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/7] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/8] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/9] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/10] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/11] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/12] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/13] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/14] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/15] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/16] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/17] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/18] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/19] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/20] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/21] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/22] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/23] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/24] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/25] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/26] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/27] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/28] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/29] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/30] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/31] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/32] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/33] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/34] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/35] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/36] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/37] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/38] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/39] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/40] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/41] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/42] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/43] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/44] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/45] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/46] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/47] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/48] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/49] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/50] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/51] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/52] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/53] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/54] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/55] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/56] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/57] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/58] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/59] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/60] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/61] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/62] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/63] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/64] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/65] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/66] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/67] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/68] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/69] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/70] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/71] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/72] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/73] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/74] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/75] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/76] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/77] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/78] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/79] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/80] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/81] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/82] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/83] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/84] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/85] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/86] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/87] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/88] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/89] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/90] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/91] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/92] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/93] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/94] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/95] 10G 9000 topology/pod-1/node-103/sys/phys-[eth1/96] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/97] inherit 9150 topology/pod-1/node-104/sys/phys-[eth1/98] inherit 9150 topology/pod-1/node-104/sys/phys-[eth1/99] inherit 9150 topology/pod-1/node-104/sys/phys-[eth1/100] inherit 9150 topology/pod-1/node-104/sys/phys-[eth1/101] inherit 9150 topology/pod-1/node-104/sys/phys-[eth1/102] inherit 9150 topology/pod-1/node-104/sys/phys-[eth1/103] inherit 9150 topology/pod-1/node-104/sys/phys-[eth1/104] inherit 9150 topology/pod-1/node-104/sys/phys-[eth1/1] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/2] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/3] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/4] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/5] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/6] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/7] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/8] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/9] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/10] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/11] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/12] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/13] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/14] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/15] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/16] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/17] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/18] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/19] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/20] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/21] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/22] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/23] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/24] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/25] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/26] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/27] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/28] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/29] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/30] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/31] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/32] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/33] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/34] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/35] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/36] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/37] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/38] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/39] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/40] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/41] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/42] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/43] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/44] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/45] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/46] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/47] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/48] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/49] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/50] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/51] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/52] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/53] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/54] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/55] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/56] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/57] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/58] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/59] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/60] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/61] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/62] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/63] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/64] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/65] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/66] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/67] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/68] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/69] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/70] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/71] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/72] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/73] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/74] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/75] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/76] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/77] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/78] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/79] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/80] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/81] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/82] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/83] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/84] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/85] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/86] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/87] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/88] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/89] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/90] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/91] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/92] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/93] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/94] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/95] 10G 9000 topology/pod-1/node-104/sys/phys-[eth1/96] 10G 9000 topology/pod-1/node-200/sys/phys-[eth1/33] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/34] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/35] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/36] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/1] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/2] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/3] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/4] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/5] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/6] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/7] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/8] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/9] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/10] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/11] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/12] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/13] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/14] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/15] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/16] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/17] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/18] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/19] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/20] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/21] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/22] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/23] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/24] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/25] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/26] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/27] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/28] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/29] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/30] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/31] inherit 9150 topology/pod-1/node-200/sys/phys-[eth1/32] inherit 9150 topology/pod-1/node-102/sys/phys-[eth1/33] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/34] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/35] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/36] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/37] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/38] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/39] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/40] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/41] inherit 9000 topology/pod-1/node-102/sys/phys-[eth1/42] inherit 9000 topology/pod-1/node-102/sys/phys-[eth1/43] inherit 9000 topology/pod-1/node-102/sys/phys-[eth1/44] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/45] 1G 9000 topology/pod-1/node-102/sys/phys-[eth1/46] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/47] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/48] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/49] inherit 9150 topology/pod-1/node-102/sys/phys-[eth1/50] inherit 9150 topology/pod-1/node-102/sys/phys-[eth1/51] inherit 9150 topology/pod-1/node-102/sys/phys-[eth1/52] inherit 9150 topology/pod-1/node-102/sys/phys-[eth1/53] inherit 9150 topology/pod-1/node-102/sys/phys-[eth1/54] inherit 9150 topology/pod-1/node-102/sys/phys-[eth1/55] inherit 9150 topology/pod-1/node-102/sys/phys-[eth1/56] inherit 9150 topology/pod-1/node-102/sys/phys-[eth1/57] inherit 9150 topology/pod-1/node-102/sys/phys-[eth1/58] inherit 9150 topology/pod-1/node-102/sys/phys-[eth1/59] inherit 9150 topology/pod-1/node-102/sys/phys-[eth1/60] inherit 9150 topology/pod-1/node-102/sys/phys-[eth1/1] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/2] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/3] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/4] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/5] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/6] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/7] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/8] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/9] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/10] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/11] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/12] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/13] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/14] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/15] 1G 9000 topology/pod-1/node-102/sys/phys-[eth1/16] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/17] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/18] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/19] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/20] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/21] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/22] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/23] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/24] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/25] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/26] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/27] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/28] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/29] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/30] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/31] 10G 9000 topology/pod-1/node-102/sys/phys-[eth1/32] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/33] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/34] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/35] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/36] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/37] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/38] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/39] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/40] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/41] inherit 9000 topology/pod-1/node-101/sys/phys-[eth1/42] inherit 9000 topology/pod-1/node-101/sys/phys-[eth1/43] inherit 9000 topology/pod-1/node-101/sys/phys-[eth1/44] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/45] 1G 9000 topology/pod-1/node-101/sys/phys-[eth1/46] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/47] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/48] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/49] inherit 9150 topology/pod-1/node-101/sys/phys-[eth1/50] inherit 9150 topology/pod-1/node-101/sys/phys-[eth1/51] inherit 9150 topology/pod-1/node-101/sys/phys-[eth1/52] inherit 9150 topology/pod-1/node-101/sys/phys-[eth1/53] inherit 9150 topology/pod-1/node-101/sys/phys-[eth1/54] inherit 9150 topology/pod-1/node-101/sys/phys-[eth1/55] inherit 9150 topology/pod-1/node-101/sys/phys-[eth1/56] inherit 9150 topology/pod-1/node-101/sys/phys-[eth1/57] inherit 9150 topology/pod-1/node-101/sys/phys-[eth1/58] inherit 9150 topology/pod-1/node-101/sys/phys-[eth1/59] inherit 9150 topology/pod-1/node-101/sys/phys-[eth1/60] inherit 9150 topology/pod-1/node-101/sys/phys-[eth1/1] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/2] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/3] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/4] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/5] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/6] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/7] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/8] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/9] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/10] 1G 9000 topology/pod-1/node-101/sys/phys-[eth1/11] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/12] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/13] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/14] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/15] 1G 9000 topology/pod-1/node-101/sys/phys-[eth1/16] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/17] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/18] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/19] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/20] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/21] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/22] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/23] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/24] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/25] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/26] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/27] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/28] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/29] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/30] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/31] 10G 9000 topology/pod-1/node-101/sys/phys-[eth1/32] 10G 9000 [None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None]
### Hints
print("DN Description Speed MTU")
print("-------------------------------------------------- -------------------- ------- ------")
print(f"{dn:50} {descr:20} {speed:7} {mtu:6}")
JSON Schema¶
[by Wikipedia] JSON Schema specifies a JSON-based format to define the structure of JSON data for validation, documentation, and interaction control. It provides a contract for the JSON data required by a given application, and how that data can be modified.
Example Schema¶
%%writefile stock.schema.json
{
"$schema": "http://json-schema.org/schema#",
"title": "Product",
"type": "object",
"required": ["id", "name", "price"],
"properties": {
"id": {
"type": "number",
"description": "Product identifier"
},
"name": {
"type": "string",
"description": "Name of the product"
},
"price": {
"type": "number",
"minimum": 0
},
"tags": {
"type": "array",
"items": {
"type": "string"
}
},
"stock": {
"type": "object",
"properties": {
"warehouse": {
"type": "number"
},
"retail": {
"type": "number"
}
}
}
}
}
Overwriting stock.schema.json
Example JSON¶
%%writefile stock.json
{
"id": 1,
"name": "Foo",
"price": 123,
"tags": [
"Bar",
"Eek"
],
"stock": {
"warehouse": 300,
"retail": 20,
"asdf": 30
}
}
Overwriting stock.json
import json
with open('stock.schema.json') as s:
schema = json.load(s)
with open('stock.json') as s:
stock = json.load(s)
schema
{'$schema': 'http://json-schema.org/schema#', 'title': 'Product', 'type': 'object', 'required': ['id', 'name', 'price'], 'properties': {'id': {'type': 'number', 'description': 'Product identifier'}, 'name': {'type': 'string', 'description': 'Name of the product'}, 'price': {'type': 'number', 'minimum': 0}, 'tags': {'type': 'array', 'items': {'type': 'string'}}, 'stock': {'type': 'object', 'properties': {'warehouse': {'type': 'number'}, 'retail': {'type': 'number'}}}}}
stock
{'id': 1, 'name': 'Foo', 'price': 123, 'tags': ['Bar', 'Eek'], 'stock': {'warehouse': 300, 'retail': 20, 'asdf': 30}}
Validation with jsonschema¶¶
! pip install jsonschema
Requirement already satisfied: jsonschema in ./.venv/lib/python3.10/site-packages (4.22.0) Requirement already satisfied: attrs>=22.2.0 in ./.venv/lib/python3.10/site-packages (from jsonschema) (23.2.0) Requirement already satisfied: jsonschema-specifications>=2023.03.6 in ./.venv/lib/python3.10/site-packages (from jsonschema) (2023.12.1) Requirement already satisfied: referencing>=0.28.4 in ./.venv/lib/python3.10/site-packages (from jsonschema) (0.35.1) Requirement already satisfied: rpds-py>=0.7.1 in ./.venv/lib/python3.10/site-packages (from jsonschema) (0.18.1) [notice] A new release of pip is available: 24.0 -> 24.1 [notice] To update, run: pip install --upgrade pip
import jsonschema
print(jsonschema.validate(stock, schema)) # ==> no error
None
invalid_stock = stock.copy()
invalid_stock.pop('id')
print(jsonschema.validate(invalid_stock, schema)) # ==> ValidationError
--------------------------------------------------------------------------- ValidationError Traceback (most recent call last) Cell In[63], line 3 1 invalid_stock = stock.copy() 2 invalid_stock.pop('id') ----> 3 print(jsonschema.validate(invalid_stock, schema)) # ==> ValidationError File ~/python_training/advanced_python/.venv/lib/python3.10/site-packages/jsonschema/validators.py:1332, in validate(instance, schema, cls, *args, **kwargs) 1330 error = exceptions.best_match(validator.iter_errors(instance)) 1331 if error is not None: -> 1332 raise error ValidationError: 'id' is a required property Failed validating 'required' in schema: {'$schema': 'http://json-schema.org/schema#', 'properties': {'id': {'description': 'Product identifier', 'type': 'number'}, 'name': {'description': 'Name of the product', 'type': 'string'}, 'price': {'minimum': 0, 'type': 'number'}, 'stock': {'properties': {'retail': {'type': 'number'}, 'warehouse': {'type': 'number'}}, 'type': 'object'}, 'tags': {'items': {'type': 'string'}, 'type': 'array'}}, 'required': ['id', 'name', 'price'], 'title': 'Product', 'type': 'object'} On instance: {'name': 'Foo', 'price': 123, 'stock': {'asdf': 30, 'retail': 20, 'warehouse': 300}, 'tags': ['Bar', 'Eek']}
Fast validation with jsonschema¶
jsonschema.Draft7Validator is a class that provides a more granular and flexible way to validate JSON data. It is specifically tailored to handle JSON Schema Draft 7. Here's a detailed explanation:
- Instantiation: You create an instance of Draft7Validator with a specific schema.
- Reusability: The created validator instance can be reused to validate multiple JSON documents against the same schema.
- Control: You have more control over the validation process, such as iterating over validation errors, checking if a JSON instance is valid, and customizing error messages.
validator = jsonschema.Draft7Validator(schema)
validator.validate(stock)
validator.validate(invalid_stock)
--------------------------------------------------------------------------- ValidationError Traceback (most recent call last) Cell In[66], line 1 ----> 1 validator.validate(invalid_stock) File ~/python_training/advanced_python/.venv/lib/python3.10/site-packages/jsonschema/validators.py:451, in create.<locals>.Validator.validate(self, *args, **kwargs) 449 def validate(self, *args, **kwargs): 450 for error in self.iter_errors(*args, **kwargs): --> 451 raise error ValidationError: 'id' is a required property Failed validating 'required' in schema: {'$schema': 'http://json-schema.org/schema#', 'properties': {'id': {'description': 'Product identifier', 'type': 'number'}, 'name': {'description': 'Name of the product', 'type': 'string'}, 'price': {'minimum': 0, 'type': 'number'}, 'stock': {'properties': {'retail': {'type': 'number'}, 'warehouse': {'type': 'number'}}, 'type': 'object'}, 'tags': {'items': {'type': 'string'}, 'type': 'array'}}, 'required': ['id', 'name', 'price'], 'title': 'Product', 'type': 'object'} On instance: {'name': 'Foo', 'price': 123, 'stock': {'asdf': 30, 'retail': 20, 'warehouse': 300}, 'tags': ['Bar', 'Eek']}
errors = list(validator.iter_errors(invalid_stock))
if errors:
for error in errors:
print(error.message)
else:
print("JSON data is valid.")
'id' is a required property
use Pydantic models for json schema validation¶
! pip install pydantic
Requirement already satisfied: pydantic in ./.venv/lib/python3.10/site-packages (2.7.4) Requirement already satisfied: annotated-types>=0.4.0 in ./.venv/lib/python3.10/site-packages (from pydantic) (0.7.0) Requirement already satisfied: pydantic-core==2.18.4 in ./.venv/lib/python3.10/site-packages (from pydantic) (2.18.4) Requirement already satisfied: typing-extensions>=4.6.1 in ./.venv/lib/python3.10/site-packages (from pydantic) (4.12.2) [notice] A new release of pip is available: 24.0 -> 24.1 [notice] To update, run: pip install --upgrade pip
! pip install datamodel-code-generator
Requirement already satisfied: datamodel-code-generator in ./.venv/lib/python3.10/site-packages (0.25.7) Requirement already satisfied: argcomplete<4.0,>=1.10 in ./.venv/lib/python3.10/site-packages (from datamodel-code-generator) (3.4.0) Requirement already satisfied: black>=19.10b0 in ./.venv/lib/python3.10/site-packages (from datamodel-code-generator) (24.4.2) Requirement already satisfied: genson<2.0,>=1.2.1 in ./.venv/lib/python3.10/site-packages (from datamodel-code-generator) (1.3.0) Requirement already satisfied: inflect<6.0,>=4.1.0 in ./.venv/lib/python3.10/site-packages (from datamodel-code-generator) (5.6.2) Requirement already satisfied: isort<6.0,>=4.3.21 in ./.venv/lib/python3.10/site-packages (from datamodel-code-generator) (5.13.2) Requirement already satisfied: jinja2<4.0,>=2.10.1 in ./.venv/lib/python3.10/site-packages (from datamodel-code-generator) (3.1.4) Requirement already satisfied: packaging in ./.venv/lib/python3.10/site-packages (from datamodel-code-generator) (24.1) Requirement already satisfied: pydantic!=2.4.0,<3.0,>=1.9.0 in ./.venv/lib/python3.10/site-packages (from pydantic[email]!=2.4.0,<3.0,>=1.9.0; python_version >= "3.10" and python_version < "3.11"->datamodel-code-generator) (2.7.4) Requirement already satisfied: pyyaml>=6.0.1 in ./.venv/lib/python3.10/site-packages (from datamodel-code-generator) (6.0.1) Requirement already satisfied: toml<1.0.0,>=0.10.0 in ./.venv/lib/python3.10/site-packages (from datamodel-code-generator) (0.10.2) Requirement already satisfied: click>=8.0.0 in ./.venv/lib/python3.10/site-packages (from black>=19.10b0->datamodel-code-generator) (8.1.7) Requirement already satisfied: mypy-extensions>=0.4.3 in ./.venv/lib/python3.10/site-packages (from black>=19.10b0->datamodel-code-generator) (1.0.0) Requirement already satisfied: pathspec>=0.9.0 in ./.venv/lib/python3.10/site-packages (from black>=19.10b0->datamodel-code-generator) (0.12.1) Requirement already satisfied: platformdirs>=2 in ./.venv/lib/python3.10/site-packages (from black>=19.10b0->datamodel-code-generator) (4.2.2) Requirement already satisfied: tomli>=1.1.0 in ./.venv/lib/python3.10/site-packages (from black>=19.10b0->datamodel-code-generator) (2.0.1) Requirement already satisfied: typing-extensions>=4.0.1 in ./.venv/lib/python3.10/site-packages (from black>=19.10b0->datamodel-code-generator) (4.12.2) Requirement already satisfied: MarkupSafe>=2.0 in ./.venv/lib/python3.10/site-packages (from jinja2<4.0,>=2.10.1->datamodel-code-generator) (2.1.5) Requirement already satisfied: annotated-types>=0.4.0 in ./.venv/lib/python3.10/site-packages (from pydantic!=2.4.0,<3.0,>=1.9.0->pydantic[email]!=2.4.0,<3.0,>=1.9.0; python_version >= "3.10" and python_version < "3.11"->datamodel-code-generator) (0.7.0) Requirement already satisfied: pydantic-core==2.18.4 in ./.venv/lib/python3.10/site-packages (from pydantic!=2.4.0,<3.0,>=1.9.0->pydantic[email]!=2.4.0,<3.0,>=1.9.0; python_version >= "3.10" and python_version < "3.11"->datamodel-code-generator) (2.18.4) Requirement already satisfied: email-validator>=2.0.0 in ./.venv/lib/python3.10/site-packages (from pydantic[email]!=2.4.0,<3.0,>=1.9.0; python_version >= "3.10" and python_version < "3.11"->datamodel-code-generator) (2.2.0) Requirement already satisfied: dnspython>=2.0.0 in ./.venv/lib/python3.10/site-packages (from email-validator>=2.0.0->pydantic[email]!=2.4.0,<3.0,>=1.9.0; python_version >= "3.10" and python_version < "3.11"->datamodel-code-generator) (2.6.1) Requirement already satisfied: idna>=2.0.0 in ./.venv/lib/python3.10/site-packages (from email-validator>=2.0.0->pydantic[email]!=2.4.0,<3.0,>=1.9.0; python_version >= "3.10" and python_version < "3.11"->datamodel-code-generator) (3.7) [notice] A new release of pip is available: 24.0 -> 24.1 [notice] To update, run: pip install --upgrade pip
! datamodel-codegen --input stock.schema.json --input-file-type jsonschema --output model.py
from model import Product
print(stock)
print(type(stock))
product_obj = Product(**stock)
print(type(product_obj))
{'id': 1, 'name': 'Foo', 'price': 123, 'tags': ['Bar', 'Eek'], 'stock': {'warehouse': 300, 'retail': 20, 'asdf': 30}} <class 'dict'> <class 'model.Product'>
print(product_obj)
id=1.0 name='Foo' price=123.0 tags=['Bar', 'Eek'] stock=Stock(warehouse=300.0, retail=20.0)
Product(**invalid_stock)
--------------------------------------------------------------------------- ValidationError Traceback (most recent call last) Cell In[80], line 1 ----> 1 Product(**invalid_stock) File ~/python_training/advanced_python/.venv/lib/python3.10/site-packages/pydantic/main.py:176, in BaseModel.__init__(self, **data) 174 # `__tracebackhide__` tells pytest and some other tools to omit this function from tracebacks 175 __tracebackhide__ = True --> 176 self.__pydantic_validator__.validate_python(data, self_instance=self) ValidationError: 1 validation error for Product id Field required [type=missing, input_value={'name': 'Foo', 'price': ...etail': 20, 'asdf': 30}}, input_type=dict] For further information visit https://errors.pydantic.dev/2.7/v/missing
Exercise: Schema for Interfaces¶
Write minimal JSON Schema for the interfaces data from the previous exercise to make sure that it will work. Enhance the solution of the previous exercise by adding validation before processing the data.
Note that mtu is saved in the JSON as a string instead of a number.
%%writefile interfaces.schema.json
{
"$schema": "http://json-schema.org/schema#",
"title": "Interfaces data",
"type": "object",
"required": ["imdata"],
"properties": {
"imdata": {
"type": "array",
"items": {
"type": "object",
"required": ["l1PhysIf"],
"properties": {
"l1PhysIf": {
"type": "object",
"required": ["attributes"],
"properties": {
"attributes": {
"type": "object",
"required": ["dn", "descr", "speed", "mtu"],
"properties": {
"dn": {"type": "string"},
"descr": {"type": "string"},
"speed": {"type": "string"},
"mtu": {"type": "string"}
}
}
}
}
}
}
}
}
}
Overwriting interfaces.schema.json
import json
import jsonschema
with open('interface-data.json') as stream:
data = json.load(stream)
with open('interfaces.schema.json') as stream:
schema = json.load(stream)
jsonschema.validate(data, schema)
Working with YAML¶
YAML Format¶
Outputing YAML¶
! pip install pyyaml
Requirement already satisfied: pyyaml in ./.venv/lib/python3.10/site-packages (6.0.1) [notice] A new release of pip is available: 24.0 -> 24.1 [notice] To update, run: pip install --upgrade pip
import yaml
with open('stock.schema.json') as stream:
schema_as_dict = json.load(stream) # create pythion dict
print(yaml.dump(schema_as_dict)) # translate pythin dict into string with yaml format
$schema: http://json-schema.org/schema# properties: id: description: Product identifier type: number name: description: Name of the product type: string price: minimum: 0 type: number stock: properties: retail: type: number warehouse: type: number type: object tags: items: type: string type: array required: - id - name - price title: Product type: object
with open('interface-data.json') as stream:
as_dict = json.load(stream)
print(yaml.dump(as_dict))
with open('interface-data.yml', 'w') as file:
file.write(yaml.dump(as_dict))
with open('test.yml', 'w') as file:
file.write(yaml.dump({"a":5}))
Loading YAML¶
string_ = '''
- my list
- qwer
- asdf
- zxcv
'''
# open from string
print(yaml.safe_load(string_))
# no yaml.loads(stream) method!
['my list', 'qwer', 'asdf', 'zxcv']
# open from file
with open('interface-data.yml') as stream:
data = yaml.safe_load(stream)
print(data)
Loading multiple YAMLs¶
string_ = '''
id: 3
name: John
language: pl
---
id: 5
name: Bob
language: no
'''
print(list(yaml.safe_load_all(string_)))
[{'id': 3, 'name': 'John', 'language': 'pl'}, {'id': 5, 'name': 'Bob', 'language': False}]
Exercise: YAML Schema for Interfaces¶
Write a snippet of code to rewrite the JSON Schema from the previous exercise into YAML. Modify the code (that prints interface reports) so that it loads the YAML schema
# SOLUTION
with open('interfaces.schema.json') as stream:
schema = json.load(stream)
with open('interfaces.schema.yaml', 'w') as stream:
stream.write(yaml.dump(schema))
import json
import jsonschema
with open('interface-data.json') as stream:
data = json.load(stream)
with open('interfaces.schema.yaml') as file:
schema = yaml.safe_load(file.read())
jsonschema.validate(data, schema)
! cat interfaces.schema.yaml
$schema: http://json-schema.org/schema# properties: imdata: items: properties: l1PhysIf: properties: attributes: properties: descr: type: string dn: type: string mtu: type: string speed: type: string required: - dn - descr - speed - mtu type: object required: - attributes type: object required: - l1PhysIf type: object type: array required: - imdata title: Interfaces data type: object
Pydantic and jsonschema¶
Example JSON¶
%%writefile data.json
{
"hasMore": false,
"participantList": [
{
"displayName": "Piotr Hyzy",
"firstName": "Piotr",
"isDeleted": false,
"lastName": "Hyzy",
"type": "REGULAR",
"userId": "35595c57-8871-44d8-9bf3-031f876c765e"
},
{
"displayName": "Piotr test",
"firstName": "Piotr",
"isDeleted": false,
"lastName": "test",
"type": "REGULAR",
"userId": "5aa65e6e-e450-4c95-839d-c8a973d18a51"
}
]
}
Overwriting data.json
Example¶
from pydantic import BaseModel
import json
class Participant(BaseModel):
displayName: str
firstName: str
isDeleted: bool
lastName: str
type: str
userId: str
class Entry(BaseModel):
hasMore: bool
participantList: list[Participant]
with open('data.json') as s:
data = json.load(s)
print(type(data)) # <class 'dict'>
<class 'dict'>
Loading from dict¶
entry = Entry(**data)
print(entry)
print(entry.hasMore)
print(entry.participantList[0])
hasMore=False participantList=[Participant(displayName='Piotr Hyzy', firstName='Piotr', isDeleted=False, lastName='Hyzy', type='REGULAR', userId='35595c57-8871-44d8-9bf3-031f876c765e'), Participant(displayName='Piotr test', firstName='Piotr', isDeleted=False, lastName='test', type='REGULAR', userId='5aa65e6e-e450-4c95-839d-c8a973d18a51')] False displayName='Piotr Hyzy' firstName='Piotr' isDeleted=False lastName='Hyzy' type='REGULAR' userId='35595c57-8871-44d8-9bf3-031f876c765e'
Generating to dict¶
entry_dict = entry.model_dump()
print(entry_dict)
print(type(entry_dict))
{'hasMore': False, 'participantList': [{'displayName': 'Piotr Hyzy', 'firstName': 'Piotr', 'isDeleted': False, 'lastName': 'Hyzy', 'type': 'REGULAR', 'userId': '35595c57-8871-44d8-9bf3-031f876c765e'}, {'displayName': 'Piotr test', 'firstName': 'Piotr', 'isDeleted': False, 'lastName': 'test', 'type': 'REGULAR', 'userId': '5aa65e6e-e450-4c95-839d-c8a973d18a51'}]} <class 'dict'>
Generating JSON Schema¶
json_schema = Entry.model_json_schema()
print(json.dumps(json_schema, indent=4))
{ "$defs": { "Participant": { "properties": { "displayName": { "title": "Displayname", "type": "string" }, "firstName": { "title": "Firstname", "type": "string" }, "isDeleted": { "title": "Isdeleted", "type": "boolean" }, "lastName": { "title": "Lastname", "type": "string" }, "type": { "title": "Type", "type": "string" }, "userId": { "title": "Userid", "type": "string" } }, "required": [ "displayName", "firstName", "isDeleted", "lastName", "type", "userId" ], "title": "Participant", "type": "object" } }, "properties": { "hasMore": { "title": "Hasmore", "type": "boolean" }, "participantList": { "items": { "$ref": "#/$defs/Participant" }, "title": "Participantlist", "type": "array" } }, "required": [ "hasMore", "participantList" ], "title": "Entry", "type": "object" }
Exercise: Interfaces Data¶
Write proper Pydantic models so that you can load info about interfaces. Then, generate the report.
### SOLUTION
import json
from pydantic import BaseModel
class PhysicalInterfaceInfo(BaseModel):
descr: str
dn: str
mtu: str
speed: str
class PhysicalInterface(BaseModel):
attributes: PhysicalInterfaceInfo
class InterfaceInfo(BaseModel):
l1PhysIf: PhysicalInterface
class Data(BaseModel):
imdata: list[InterfaceInfo]
# Load JSON data from file
with open('interface-data.json') as stream:
data_as_dict = json.load(stream)
# Parse JSON data into Pydantic model
data = Data(**data_as_dict)
# Generate the report
print("DN Description Speed MTU")
print("-------------------------------------------------- -------------------- ------- ------")
for interface in data.imdata[:10]:
dn = interface.l1PhysIf.attributes.dn
descr = interface.l1PhysIf.attributes.descr
speed = interface.l1PhysIf.attributes.speed
mtu = interface.l1PhysIf.attributes.mtu
# print fields formatted in columns
print(f"{dn:50} {descr:20} {speed:7} {mtu:6}")
DN Description Speed MTU -------------------------------------------------- -------------------- ------- ------ topology/pod-1/node-201/sys/phys-[eth1/33] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/34] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/35] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/36] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/1] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/2] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/3] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/4] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/5] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/6] inherit 9150
Data.model_json_schema()
{'$defs': {'InterfaceInfo': {'properties': {'l1PhysIf': {'$ref': '#/$defs/PhysicalInterface'}}, 'required': ['l1PhysIf'], 'title': 'InterfaceInfo', 'type': 'object'}, 'PhysicalInterface': {'properties': {'attributes': {'$ref': '#/$defs/PhysicalInterfaceInfo'}}, 'required': ['attributes'], 'title': 'PhysicalInterface', 'type': 'object'}, 'PhysicalInterfaceInfo': {'properties': {'descr': {'title': 'Descr', 'type': 'string'}, 'dn': {'title': 'Dn', 'type': 'string'}, 'mtu': {'title': 'Mtu', 'type': 'string'}, 'speed': {'title': 'Speed', 'type': 'string'}}, 'required': ['descr', 'dn', 'mtu', 'speed'], 'title': 'PhysicalInterfaceInfo', 'type': 'object'}}, 'properties': {'imdata': {'items': {'$ref': '#/$defs/InterfaceInfo'}, 'title': 'Imdata', 'type': 'array'}}, 'required': ['imdata'], 'title': 'Data', 'type': 'object'}
### SOLUTION
import json
from pydantic import BaseModel
class PhysicalInterfaceInfo(BaseModel):
descr: str
dn: str
mtu: str
speed: str
class PhysicalInterface(BaseModel):
attributes: PhysicalInterfaceInfo
class InterfaceInfo(BaseModel):
l1PhysIf: PhysicalInterface
@property
def attr(self):
return self.l1PhysIf.attributes
def __str__(self):
return f"{self.attr.dn:50} {self.attr.descr:20} {self.attr.speed:7} {self.attr.mtu:6}"
class Data(BaseModel):
imdata: list[InterfaceInfo]
def print_report(self):
print("DN Description Speed MTU")
print("-------------------------------------------------- -------------------- ------- ------")
_ = [print(_interface) for _interface in self.imdata[:10]]
# Load JSON data from file
with open('interface-data.json') as stream:
data_as_dict = json.load(stream)
# Parse JSON data into Pydantic model
data = Data(**data_as_dict)
data.print_report()
DN Description Speed MTU -------------------------------------------------- -------------------- ------- ------ topology/pod-1/node-201/sys/phys-[eth1/33] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/34] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/35] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/36] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/1] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/2] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/3] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/4] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/5] inherit 9150 topology/pod-1/node-201/sys/phys-[eth1/6] inherit 9150
Explanation
- Pydantic Model Definition: The
PhysicalInterfaceInfo
,PhysicalInterface
,InterfaceInfo
, andData
models are defined to match the structure of the JSON data. - Loading JSON Data: The JSON data is loaded from the
interface-data.json
file. - Parsing Data: The JSON data is parsed into an instance of the
Data
model using the**data_as_dict
syntax. - Generating the Report: The report is generated by iterating over the first 10 interfaces and printing their details in a formatted manner.
Creating REST APIs with FastAPI and Pydantic¶
REST API Principles¶
Example API¶
! pip install fastapi
Requirement already satisfied: fastapi in ./.venv/lib/python3.10/site-packages (0.111.0) Requirement already satisfied: starlette<0.38.0,>=0.37.2 in ./.venv/lib/python3.10/site-packages (from fastapi) (0.37.2) Requirement already satisfied: pydantic!=1.8,!=1.8.1,!=2.0.0,!=2.0.1,!=2.1.0,<3.0.0,>=1.7.4 in ./.venv/lib/python3.10/site-packages (from fastapi) (2.7.4) Requirement already satisfied: typing-extensions>=4.8.0 in ./.venv/lib/python3.10/site-packages (from fastapi) (4.12.2) Requirement already satisfied: fastapi-cli>=0.0.2 in ./.venv/lib/python3.10/site-packages (from fastapi) (0.0.4) Requirement already satisfied: httpx>=0.23.0 in ./.venv/lib/python3.10/site-packages (from fastapi) (0.27.0) Requirement already satisfied: jinja2>=2.11.2 in ./.venv/lib/python3.10/site-packages (from fastapi) (3.1.4) Requirement already satisfied: python-multipart>=0.0.7 in ./.venv/lib/python3.10/site-packages (from fastapi) (0.0.9) Requirement already satisfied: ujson!=4.0.2,!=4.1.0,!=4.2.0,!=4.3.0,!=5.0.0,!=5.1.0,>=4.0.1 in ./.venv/lib/python3.10/site-packages (from fastapi) (5.10.0) Requirement already satisfied: orjson>=3.2.1 in ./.venv/lib/python3.10/site-packages (from fastapi) (3.10.5) Requirement already satisfied: email_validator>=2.0.0 in ./.venv/lib/python3.10/site-packages (from fastapi) (2.2.0) Requirement already satisfied: uvicorn>=0.12.0 in ./.venv/lib/python3.10/site-packages (from uvicorn[standard]>=0.12.0->fastapi) (0.30.1) Requirement already satisfied: dnspython>=2.0.0 in ./.venv/lib/python3.10/site-packages (from email_validator>=2.0.0->fastapi) (2.6.1) Requirement already satisfied: idna>=2.0.0 in ./.venv/lib/python3.10/site-packages (from email_validator>=2.0.0->fastapi) (3.7) Requirement already satisfied: typer>=0.12.3 in ./.venv/lib/python3.10/site-packages (from fastapi-cli>=0.0.2->fastapi) (0.12.3) Requirement already satisfied: anyio in ./.venv/lib/python3.10/site-packages (from httpx>=0.23.0->fastapi) (4.4.0) Requirement already satisfied: certifi in ./.venv/lib/python3.10/site-packages (from httpx>=0.23.0->fastapi) (2024.6.2) Requirement already satisfied: httpcore==1.* in ./.venv/lib/python3.10/site-packages (from httpx>=0.23.0->fastapi) (1.0.5) Requirement already satisfied: sniffio in ./.venv/lib/python3.10/site-packages (from httpx>=0.23.0->fastapi) (1.3.1) Requirement already satisfied: h11<0.15,>=0.13 in ./.venv/lib/python3.10/site-packages (from httpcore==1.*->httpx>=0.23.0->fastapi) (0.14.0) Requirement already satisfied: MarkupSafe>=2.0 in ./.venv/lib/python3.10/site-packages (from jinja2>=2.11.2->fastapi) (2.1.5) Requirement already satisfied: annotated-types>=0.4.0 in ./.venv/lib/python3.10/site-packages (from pydantic!=1.8,!=1.8.1,!=2.0.0,!=2.0.1,!=2.1.0,<3.0.0,>=1.7.4->fastapi) (0.7.0) Requirement already satisfied: pydantic-core==2.18.4 in ./.venv/lib/python3.10/site-packages (from pydantic!=1.8,!=1.8.1,!=2.0.0,!=2.0.1,!=2.1.0,<3.0.0,>=1.7.4->fastapi) (2.18.4) Requirement already satisfied: click>=7.0 in ./.venv/lib/python3.10/site-packages (from uvicorn>=0.12.0->uvicorn[standard]>=0.12.0->fastapi) (8.1.7) Requirement already satisfied: httptools>=0.5.0 in ./.venv/lib/python3.10/site-packages (from uvicorn[standard]>=0.12.0->fastapi) (0.6.1) Requirement already satisfied: python-dotenv>=0.13 in ./.venv/lib/python3.10/site-packages (from uvicorn[standard]>=0.12.0->fastapi) (1.0.1) Requirement already satisfied: pyyaml>=5.1 in ./.venv/lib/python3.10/site-packages (from uvicorn[standard]>=0.12.0->fastapi) (6.0.1) Requirement already satisfied: uvloop!=0.15.0,!=0.15.1,>=0.14.0 in ./.venv/lib/python3.10/site-packages (from uvicorn[standard]>=0.12.0->fastapi) (0.19.0) Requirement already satisfied: watchfiles>=0.13 in ./.venv/lib/python3.10/site-packages (from uvicorn[standard]>=0.12.0->fastapi) (0.22.0) Requirement already satisfied: websockets>=10.4 in ./.venv/lib/python3.10/site-packages (from uvicorn[standard]>=0.12.0->fastapi) (12.0) Requirement already satisfied: exceptiongroup>=1.0.2 in ./.venv/lib/python3.10/site-packages (from anyio->httpx>=0.23.0->fastapi) (1.2.1) Requirement already satisfied: shellingham>=1.3.0 in ./.venv/lib/python3.10/site-packages (from typer>=0.12.3->fastapi-cli>=0.0.2->fastapi) (1.5.4) Requirement already satisfied: rich>=10.11.0 in ./.venv/lib/python3.10/site-packages (from typer>=0.12.3->fastapi-cli>=0.0.2->fastapi) (13.7.1) Requirement already satisfied: markdown-it-py>=2.2.0 in ./.venv/lib/python3.10/site-packages (from rich>=10.11.0->typer>=0.12.3->fastapi-cli>=0.0.2->fastapi) (3.0.0) Requirement already satisfied: pygments<3.0.0,>=2.13.0 in ./.venv/lib/python3.10/site-packages (from rich>=10.11.0->typer>=0.12.3->fastapi-cli>=0.0.2->fastapi) (2.18.0) Requirement already satisfied: mdurl~=0.1 in ./.venv/lib/python3.10/site-packages (from markdown-it-py>=2.2.0->rich>=10.11.0->typer>=0.12.3->fastapi-cli>=0.0.2->fastapi) (0.1.2) [notice] A new release of pip is available: 24.0 -> 24.1 [notice] To update, run: pip install --upgrade pip
%%writefile "simple_api.py"
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
class TodoItem(BaseModel):
id: int
task: str
todos = [
TodoItem(id=1, task="build an API"),
TodoItem(id=2, task="something else"),
TodoItem(id=3, task="bla bla"),
]
@app.get("/todos", response_model=list[TodoItem]) # expose interface for data collection
def get_todos():
return todos
@app.get("/todos/{todo_id}", response_model=TodoItem)
def read_todo(todo_id: int):
for todo in todos:
if todo.id == todo_id:
return todo
raise HTTPException(status_code=404, detail="Todo not found")
@app.get("/test_return_string", response_model=str)
def get_string():
return "this is string"
if __name__ == '__main__': # secure against running a server unintentionally while importing the file
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Overwriting simple_api.py
Launch Server¶
! uvicorn simple_api:app --reload
Accessing the API with Postman¶
API documentation¶
More Methods¶
# %%writefile "simple_api.py"
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
app = FastAPI()
class TodoItem(BaseModel):
id: int
task: str
class TodoItemCreate(BaseModel):
task: str = Field(..., min_length=5) ## min length
todos = [
TodoItem(id=1, task="build an API"),
TodoItem(id=2, task="something else"),
TodoItem(id=3, task="bla bla"),
]
@app.get("/todos", response_model=list[TodoItem])
def get_todos():
return todos
@app.get("/todos/{todo_id}", response_model=TodoItem)
def read_todo(todo_id: int):
for todo in todos:
if todo.id == todo_id:
return todo
raise HTTPException(status_code=404, detail="Todo not found")
@app.post("/todos", response_model=TodoItem, status_code=201) # create a new objects, other methods which are not idempotent
def create_todo(todo: TodoItemCreate):
new_id = max(todo.id for todo in todos) + 1 if todos else 1
new_todo = TodoItem(id=new_id, task=todo.task)
todos.append(new_todo)
return new_todo
@app.put("/todos/{todo_id}", response_model=TodoItem) # idempotent update
def update_todo(todo_id: int, updated_todo: TodoItemCreate):
for todo in todos:
if todo.id == todo_id:
todo.task = updated_todo.task
return todo
raise HTTPException(status_code=404, detail="Todo not found")
@app.delete("/todos/{todo_id}", status_code=204) # removal
def delete_todo(todo_id: int):
for index, todo in enumerate(todos):
if todo.id == todo_id:
del todos[index]
return
raise HTTPException(status_code=404, detail="Todo not found")
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Overwriting simple_api.py
Exercise: Enhance the Todo API with Additional Features¶
Task:¶
Extend the provided Todo API with the following features:
- Add a new endpoint to mark a todo item as completed.
- Add a new endpoint to list only the completed todo items.
Instructions:¶
Mark Todo as Completed:
- Create a new endpoint
PUT /todos/{todo_id}/complete
that will update the status of a todo item to completed. - The todo item should have an additional attribute
completed: bool
which defaults toFalse
.
- Create a new endpoint
List Completed Todos:
- Create a new endpoint
GET /todos/completed
that will return a list of all completed todo items.
- Create a new endpoint
Example:¶
After completing the exercise, your API should support the following new functionality:
Marking a todo item as completed:
PUT /todos/1/complete
Listing all completed todo items:
GET /todos/completed
### SOLUTION
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
class TodoItem(BaseModel):
id: int
task: str
completed: bool = False
class TodoItemCreate(BaseModel):
task: str = Field(..., min_length=5) ## min length
todos = [
TodoItem(id=1, task="build an API"),
TodoItem(id=2, task="something else"),
TodoItem(id=3, task="bla bla"),
]
@app.get("/todos", response_model=list[TodoItem])
def get_todos():
return todos
@app.post("/todos", response_model=TodoItem, status_code=201)
def create_todo(todo: TodoItemCreate):
new_id = max(todo.id for todo in todos) + 1 if todos else 1
new_todo = TodoItem(id=new_id, task=todo.task)
todos.append(new_todo)
return new_todo
# 2. List Completed Todos
@app.get("/todos/completed", response_model=list[TodoItem])
def get_completed_todos():
return [todo for todo in todos if todo.completed]
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
@app.get("/todos/{todo_id}", response_model=TodoItem)
def read_todo(todo_id: int):
for todo in todos:
if todo.id == todo_id:
return todo
raise HTTPException(status_code=404, detail="Todo not found")
@app.put("/todos/{todo_id}", response_model=TodoItem)
def update_todo(todo_id: int, updated_todo: TodoItemCreate):
for todo in todos:
if todo.id == todo_id:
todo.task = updated_todo.task
return todo
raise HTTPException(status_code=404, detail="Todo not found")
@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo(todo_id: int):
for index, todo in enumerate(todos):
if todo.id == todo_id:
del todos[index]
return
raise HTTPException(status_code=404, detail="Todo not found")
# Exercise starts here
# 1. Mark Todo as Completed
@app.put("/todos/{todo_id}/complete", response_model=TodoItem)
def complete_todo(todo_id: int):
for todo in todos:
if todo.id == todo_id:
todo.completed = True
return todo
raise HTTPException(status_code=404, detail="Todo not found")
Requests Fundamentals¶
Get Method¶
# pip install requests # Installation
import requests
URL = 'http://127.0.0.1:8000'
import json
response = requests.get(f'{URL}/todos')
print(type(response.content.decode('utf-8')))
print(response.content.decode('utf-8'))
print()
print(json.loads(response.content.decode('utf-8')))
print(response.json()) # simple alternative for json.loads(response.content.decode('utf-8'))
print(type(response.json()))
print(response.status_code)
<class 'str'> [{"id":1,"task":"build an API","completed":false},{"id":2,"task":"something else","completed":false},{"id":3,"task":"bla bla","completed":false},{"id":4,"task":"umyj naczynia !","completed":false},{"id":5,"task":"wynies smieci !","completed":false},{"id":6,"task":"umyj zeby !","completed":false}] [{'id': 1, 'task': 'build an API', 'completed': False}, {'id': 2, 'task': 'something else', 'completed': False}, {'id': 3, 'task': 'bla bla', 'completed': False}, {'id': 4, 'task': 'umyj naczynia !', 'completed': False}, {'id': 5, 'task': 'wynies smieci !', 'completed': False}, {'id': 6, 'task': 'umyj zeby !', 'completed': False}] [{'id': 1, 'task': 'build an API', 'completed': False}, {'id': 2, 'task': 'something else', 'completed': False}, {'id': 3, 'task': 'bla bla', 'completed': False}, {'id': 4, 'task': 'umyj naczynia !', 'completed': False}, {'id': 5, 'task': 'wynies smieci !', 'completed': False}, {'id': 6, 'task': 'umyj zeby !', 'completed': False}] <class 'list'> 200
Post Method¶
data = {'task': 'wynies smieci !'}
response = requests.post(f'{URL}/todos', json=data)
print(response.json())
print(response.status_code)
{'id': 5, 'task': 'wynies smieci !', 'completed': False} 201
data = {'task': 'umyj zeby !'}
response = requests.post(f'{URL}/todos', data=json.dumps(data))
print(response.json())
print(response.status_code)
{'id': 6, 'task': 'umyj zeby !', 'completed': False} 201
attach params(query) to a request¶
params1 = {
'key': 'value',
'query': 'name=Piotr;surname=Hyzy',
'maxSize': 50,
}
params2 = dict(
key=5,
)
response = requests.get(f'{URL}/todos', params=params1)
response = requests.get(f'{URL}/todos', params=params2)
set custom Headers¶
headers = {'User-Agent': 'my-custom-app/1.0'}
response = requests.get(f'{URL}/todos', headers=headers)
Invalid post request¶
data = {'task': ''}
response = requests.post(f'{URL}/todos', json=data)
print(response.status_code)
print(response.content)
print(response.json())
422 b'{"detail":[{"type":"string_too_short","loc":["body","task"],"msg":"String should have at least 5 characters","input":"","ctx":{"min_length":5}}]}' {'detail': [{'type': 'string_too_short', 'loc': ['body', 'task'], 'msg': 'String should have at least 5 characters', 'input': '', 'ctx': {'min_length': 5}}]}
Put Method¶
data = {'task': 'Modified task'}
response = requests.put(f'{URL}/todos/3', json=data)
print(response.json())
{'id': 3, 'task': 'Modified task', 'completed': False}
response = requests.get(f'{URL}/todos')
print(response.json())
[{'id': 1, 'task': 'build an API', 'completed': False}, {'id': 2, 'task': 'something else', 'completed': False}, {'id': 3, 'task': 'Modified task', 'completed': False}]
Delete Method¶
response = requests.delete(f'{URL}/todos/3')
print(response.status_code)
204
response = requests.get(f'{URL}/todos')
print(response.json())
[{'id': 1, 'task': 'build an API', 'completed': False}, {'id': 2, 'task': 'something else', 'completed': False}]
Simple GUI with Pydantic¶
Intro¶
! mkdir -p fastapi-gui/
! mkdir -p fastapi-gui/templates
Templates¶
%%writefile fastapi-gui/templates/base.html
<!doctype html>
<html lang="en">
<head>
<!-- Required meta tags -->
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no">
<!-- Bootstrap CSS -->
<link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.4.1/css/bootstrap.min.css" integrity="sha384-Vkoo8x4CGsO3+Hhxv8T/Q5PaXtkKtu6ug5TOeNV6gBiFeWPGFN9MuhOf23Q9Ifjh" crossorigin="anonymous">
<title>Hello, world!</title>
</head>
<body class="bg-light">
<div class="container">
{% block content %}{% endblock %}
</div>
<!-- Optional JavaScript -->
<!-- jQuery first, then Popper.js, then Bootstrap JS -->
<script src="https://code.jquery.com/jquery-3.4.1.slim.min.js" integrity="sha384-J6qa4849blE2+poT4WnyKhv5vZF5SrPo0iEjwBvKU7imGFAV0wwj1yYfoRSJoZ+n" crossorigin="anonymous"></script>
<script src="https://cdn.jsdelivr.net/npm/popper.js@1.16.0/dist/umd/popper.min.js" integrity="sha384-Q6E9RHvbIyZFJoft+2mJbHaEWldlvI9IOYy5n3zV9zzTtmI3UksdQRVvoxMfooAo" crossorigin="anonymous"></script>
<script src="https://stackpath.bootstrapcdn.com/bootstrap/4.4.1/js/bootstrap.min.js" integrity="sha384-wfSDF2E50Y2D1uUdj0O3uMBJnjuUD4Ih7YwaYd1iqfktj0Uod8GCExl3Og8ifwB6" crossorigin="anonymous"></script>
</body>
</html>
Overwriting fastapi-gui/templates/base.html
%%writefile fastapi-gui/templates/main.html
{% extends "base.html" %}
{% block content %}
<h1>Hello world!</h1>
<form action="/run" method="POST">
<div class="form-group">
<label for="exampleFormControlInput1">Example text</label>
<input type="text" class="form-control" id="exampleFormControlInput1" name="text" />
</div>
<div class="form-group">
<label for="exampleFormControlSelect1">Example select</label>
<select class="form-control" id="exampleFormControlSelect1" name="op">
<option value="whitespace">count whitespaces</option>
<option value="all">count all characters</option>
<option value="split">split by whitespaces</option>
</select>
</div>
<button type="submit" class="btn btn-primary btn-block">Start</button>
</form>
{% endblock %}
Overwriting fastapi-gui/templates/main.html
%%writefile fastapi-gui/templates/run.html
{% extends "base.html" %}
{% block content %}
<h1>Here are the results</h1>
<div class="alert alert-success" role="alert">
Successfully finished!
</div>
<pre>
{{ msg }}
</pre>
<a href="/">Try again</a>
{% endblock %}
Overwriting fastapi-gui/templates/run.html
%%writefile fastapi-gui/templates/split.html
{% extends "base.html" %}
{% block content %}
<h1>Here are the results</h1>
<div class="alert alert-success" role="alert">
Successfully finished!
</div>
<ul>
{% for word in words %}
<li>{{ word }}</li>
{% endfor %}
</ul>
<a href="/">Try again</a>
{% endblock %}
Overwriting fastapi-gui/templates/split.html
App¶
%%writefile fastapi-gui/app.py
import webbrowser
from fastapi import FastAPI, Form, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
# from fastapi.staticfiles import StaticFiles
app = FastAPI()
# app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")
@app.get("/", response_class=HTMLResponse)
def read_root(request: Request):
return templates.TemplateResponse("main.html", {"request": request})
@app.post("/run", response_class=HTMLResponse)
def run(request: Request, text: str = Form(...), op: str = Form(...)):
if op == 'whitespace':
result = text.count(' ')
msg = f"Number of matching characters: {result}"
return templates.TemplateResponse("run.html", {"request": request, "msg": msg})
elif op == 'all':
result = len(text)
msg = f"Number of matching characters: {result}"
return templates.TemplateResponse("run.html", {"request": request, "msg": msg})
elif op == 'split':
result = text.split()
return templates.TemplateResponse("split.html", {"request": request, "words": result})
if __name__ == '__main__':
import uvicorn
webbrowser.open_new("http://127.0.0.1:8000/")
uvicorn.run(app, host="0.0.0.0", port=8000)
Overwriting fastapi-gui/app.py
Explanation¶
import webbrowser
- This imports the
webbrowser
module, which can be used to open web browsers and display web pages.
from fastapi import FastAPI, Form, Request
- This imports the
FastAPI
class for creating the web application instance,Form
for handling form data in HTTP requests, andRequest
for getting the request object in route handlers.
from fastapi.responses import HTMLResponse
- This imports the
HTMLResponse
class, which is used to return HTML content from a route.
from fastapi.templating import Jinja2Templates
- This imports the
Jinja2Templates
class, which is used to load and render Jinja2 templates.
app = FastAPI()
- This creates an instance of the FastAPI class, which represents the web application.
templates = Jinja2Templates(directory="templates")
- This creates an instance of
Jinja2Templates
configured to look for templates in thetemplates
directory.
@app.get("/", response_class=HTMLResponse)
def read_root(request: Request):
return templates.TemplateResponse("main.html", {"request": request})
- This defines a route handler for the root URL ("/"). It uses the
@app.get
decorator to specify that this route should handle GET requests. Theresponse_class
parameter specifies that the response should be an HTML document. The functionread_root
renders and returns themain.html
template, passing the request object to the template.
@app.post("/run", response_class=HTMLResponse)
def run(request: Request, text: str = Form(...), op: str = Form(...)):
if op == 'whitespace':
result = text.count(' ')
msg = f"Number of matching characters: {result}"
return templates.TemplateResponse("run.html", {"request": request, "msg": msg})
elif op == 'all':
result = len(text)
msg = f"Number of matching characters: {result}"
return templates.TemplateResponse("run.html", {"request": request, "msg": msg})
elif op == 'split':
result = text.split()
return templates.TemplateResponse("split.html", {"request": request, "words": result})
This defines a route handler for the
/run
URL. It uses the@app.post
decorator to specify that this route should handle POST requests. Theresponse_class
parameter specifies that the response should be an HTML document. The functionrun
takes the request object,text
, andop
as parameters. Thetext
andop
parameters are extracted from the form data in the POST request.- If
op
is 'whitespace', it counts the number of spaces intext
and returns therun.html
template with a message containing the result. - If
op
is 'all', it counts the total number of characters intext
and returns therun.html
template with a message containing the result. - If
op
is 'split', it splitstext
by whitespace and returns thesplit.html
template with the resulting words.
- If
if __name__ == '__main__':
import uvicorn
webbrowser.open_new("http://127.0.0.1:8000/")
uvicorn.run(app, host="0.0.0.0", port=8000)
This block runs the application if the script is executed directly (as opposed to being imported as a module).
- It imports the
uvicorn
module, which is an ASGI server for running the FastAPI application. - It opens the default web browser and navigates to
http://127.0.0.1:8000/
. - It starts the FastAPI application using
uvicorn
, making it accessible athttp://0.0.0.0:8000/
.
- It imports the
Summary¶
- The script sets up a FastAPI application with routes for displaying a form (
/
) and processing the form data (/run
). - It uses Jinja2 templates to render HTML responses.
- Static files are served from a designated directory.
- When run directly, it automatically opens the web browser to the application's URL and starts the FastAPI server.
Compiling into an Executable¶
! pip install pyinstaller
Requirement already satisfied: pyinstaller in ./.venv/lib/python3.10/site-packages (6.8.0) Requirement already satisfied: setuptools>=42.0.0 in ./.venv/lib/python3.10/site-packages (from pyinstaller) (65.5.0) Requirement already satisfied: altgraph in ./.venv/lib/python3.10/site-packages (from pyinstaller) (0.17.4) Requirement already satisfied: pyinstaller-hooks-contrib>=2024.6 in ./.venv/lib/python3.10/site-packages (from pyinstaller) (2024.7) Requirement already satisfied: packaging>=22.0 in ./.venv/lib/python3.10/site-packages (from pyinstaller) (24.1) Requirement already satisfied: macholib>=1.8 in ./.venv/lib/python3.10/site-packages (from pyinstaller) (1.16.3) [notice] A new release of pip is available: 24.0 -> 24.1 [notice] To update, run: pip install --upgrade pip
! pyinstaller app.py --onefile --noconsole --add-data "templates:templates"
Exercise: split¶
Add fourth operation: count all characters except whitespaces
Parallel Programming¶
Threading¶
The threading
module provides a way to create and manage threads, which allow for concurrent execution within a single process. Threads share the same memory space and are more lightweight than processes. However, Python's Global Interpreter Lock (GIL) means that only one thread can execute Python bytecode at a time, limiting the effectiveness of threading for CPU-bound tasks.
Key Features of threading
:
- Threads: Creates multiple threads within the same process.
- Shared Memory: Threads share the same memory space.
- Lightweight: Less overhead compared to processes.
- GIL Limitation: Due to the GIL, threading is not ideal for CPU-bound tasks but can be effective for I/O-bound tasks.
- Simpler Syntax: Easier to implement compared to multiprocessing.
Example¶
import threading
import time
def worker(delay):
for i in range(5):
time.sleep(delay)
print("Thread {:.3f}".format(time.time()))
w1 = threading.Thread(target=worker, args=[1.0]) # create new thread
w1.start() # start thread
for i in range(5):
print("Main {:.3f}".format(time.time())) # ==> wypisanie argument, wypisanie "\n"
time.sleep(0.5)
w1.join() # waith for thread completion
Main 1719306317.721 Main 1719306318.224 Thread 1719306318.725 Main 1719306318.728 Main 1719306319.233 Thread 1719306319.731 Main 1719306319.738 Thread 1719306320.736 Thread 1719306321.741 Thread 1719306322.746
import threading
import time
def print_(msg):
print('{:6.3f}: {}'.format(time.time(), msg), flush=True)
class MyThread(threading.Thread): # definition of a new thread
def __init__(self, delay):
super().__init__()
self.delay = delay # new atrtribute for MyThread
def run(self):# equivalent to target=
print_("Starting")
time.sleep(self.delay)
print_("Finishing the background thread")
thread = MyThread(2.5) # Create new Thread
thread.start() # Start new Thread
print_("Finishing the main thread")
thread.join() # wait for thread completion
1719306546.756: Finishing the main thread
Produce and Consume¶
Create one producent for each URL. Create only one consument. Make sure each producent and the consument work in their own threads.
from queue import Queue
from urllib.request import urlopen
from threading import Thread
queue = Queue() # queue for inter threads communication
def produce(url):
res = urlopen(url) # open html url and return
html = res.read() # return html content
# html = "pobrano " + url # mock
queue.put(html)
def consume():
counter = 0
while True:
html = queue.get()
time.sleep(0.1)
counter += len(html)
print('Sum of bytes is {}'.format(counter))
queue.task_done()
urls = [
'http://o2.pl',
'http://wp.pl',
]
# dwóch producentów, każdy w osobnym wątku
threads = []
for url in urls:
t = Thread(target=produce, args=[url])
t.start()
threads.append(t)
# jeden konsument, w osobnym wÄ…tku
c = Thread(target=consume, daemon=True)
c.start()
for t in threads:
t.join()
queue.join() # blocks until all elements from queue are processed (that is, queue.task_done() was called)
Sum of bytes is 207442 Sum of bytes is 1013376
Exercise: Multithreaded File Processing System¶
In this exercise, you will create a multithreaded file processing system using Python's threading
and queue
modules. The objective is to simulate a scenario where multiple threads read data from a set of files, process the data, and aggregate the results using a queue.
Requirements:
Producers (File Readers):
- Multiple threads will read data from different files concurrently.
- Each thread will simulate reading from a file and place the data into a queue.
- Use a delay to simulate the time taken to read each file.
Consumer (Data Processor):
- A single thread will read data from the queue.
- The consumer will process the data (e.g., count the number of lines) and keep an aggregate count of all lines read.
- The consumer will print the aggregate result after processing each chunk of data.
Implementation Steps:
Create Producer Threads:
- Define a
producer
function that takes a file path as an argument, reads the file, and puts the content into a queue. - Create multiple producer threads, each reading from a different file.
- Define a
Create Consumer Thread:
- Define a
consumer
function that continuously reads data from the queue, processes it, and updates the aggregate count. - The consumer should use a delay to simulate processing time.
- Define a
Main Function:
- Initialize a queue.
- Start multiple producer threads.
- Start the consumer thread.
- Ensure the main thread waits for all producer threads to finish and the queue to be fully processed.
# Step 1: Create sample text files
file_contents = [
"""Line 1 of file 1
Line 2 of file 1
Line 3 of file 1""",
"""Line 1 of file 2
Line 2 of file 2
Line 3 of file 2
Line 4 of file 2""",
"""Line 1 of file 3
Line 2 of file 3"""
]
file_names = ['file1.txt', 'file2.txt', 'file3.txt']
for file_name, content in zip(file_names, file_contents):
with open(file_name, 'w') as f:
f.write(content)
# Verification
for file_name in file_names:
with open(file_name, 'r') as f:
print(f"Contents of {file_name}:\n{f.read()}\n")
Contents of file1.txt: Line 1 of file 1 Line 2 of file 1 Line 3 of file 1 Contents of file2.txt: Line 1 of file 2 Line 2 of file 2 Line 3 of file 2 Line 4 of file 2 Contents of file3.txt: Line 1 of file 3 Line 2 of file 3
# SOLUTION
import threading
import time
from queue import Queue
# Define the producer function
def producer(file_path):
with open(file_path, 'r') as file:
data = file.read()
queue.put(data)
time.sleep(1) # Simulate delay
# Define the consumer function
def consumer():
line_count = 0
while True:
data = queue.get()
lines = data.split('\n')
line_count += len(lines)
print('Total lines processed: {}'.format(line_count))
queue.task_done()
time.sleep(0.5) # Simulate processing time
# Main function to set up threads and queue
if __name__ == "__main__":
queue = Queue()
# List of files to be processed
files = ['file1.txt', 'file2.txt', 'file3.txt']
# Create and start producer threads
producer_threads = []
for file in files:
t = threading.Thread(target=producer, args=(file,))
t.start()
producer_threads.append(t)
# Create and start consumer thread
consumer_thread = threading.Thread(target=consumer, daemon=True)
consumer_thread.start()
# Wait for all producer threads to finish
for t in producer_threads:
t.join()
# Wait for the queue to be fully processed
queue.join()
print("All files processed.")
Total lines processed: 3 Total lines processed: 7 Total lines processed: 9 All files processed.
Concurrent¶
The concurrent
module (specifically concurrent.futures
) provides a higher-level interface for asynchronously executing callables. It supports both threads and processes via ThreadPoolExecutor
and ProcessPoolExecutor
.
Key Features of concurrent.futures
:
- ThreadPoolExecutor: Manages a pool of threads for concurrent execution.
- ProcessPoolExecutor: Manages a pool of processes, bypassing the GIL and allowing for true parallelism.
- Futures: Represents a result of an asynchronous computation.
- Higher-Level Abstraction: Provides a more abstract and easier-to-use API for managing concurrency.
%%time
from time import sleep
import random
def worker(name):
sleep(0.5+random.random())
print(f"Inside {name}")
return f"The result of {name}"
names = [f"Job {i}" for i in range(1,10)]
print(names)
list(map(worker, names))
['Job 1', 'Job 2', 'Job 3', 'Job 4', 'Job 5', 'Job 6', 'Job 7', 'Job 8', 'Job 9'] Inside Job 1 Inside Job 2 Inside Job 3 Inside Job 4 Inside Job 5 Inside Job 6 Inside Job 7 Inside Job 8 Inside Job 9 CPU times: user 8.92 ms, sys: 3.21 ms, total: 12.1 ms Wall time: 9.43 s
['The result of Job 1', 'The result of Job 2', 'The result of Job 3', 'The result of Job 4', 'The result of Job 5', 'The result of Job 6', 'The result of Job 7', 'The result of Job 8', 'The result of Job 9']
Parallel Example¶
# %%time
from concurrent import futures
from time import sleep
import random
def worker(name):
sleep(0.5+random.random())
print(f"Inside {name}")
return f"The result of {name}"
names = [f"Job {i}" for i in range(1,10)]
print(names)
# list(map(worker, names))
with futures.ThreadPoolExecutor(max_workers=4) as pool: # max_workers how many threads I wish my pool manages
results = pool.map(worker, names)
for res in results:
print(res)
['Job 1', 'Job 2', 'Job 3', 'Job 4', 'Job 5', 'Job 6', 'Job 7', 'Job 8', 'Job 9'] Inside Job 1 The result of Job 1 Inside Job 4 Inside Job 3 Inside Job 2 The result of Job 2 The result of Job 3 The result of Job 4 Inside Job 7 Inside Job 6 Inside Job 5 The result of Job 5 The result of Job 6 The result of Job 7 Inside Job 8 The result of Job 8 Inside Job 9 The result of Job 9 CPU times: user 8.92 ms, sys: 3.67 ms, total: 12.6 ms Wall time: 2.48 s
Exercise: Produce & Consume¶
Implement the Produce and Consume snippet (with urls) with use concurrency module
from concurrent import futures
from urllib.request import urlopen
from time import sleep
urls: list[str] = [
'http://o2.pl',
'http://wp.pl',
]
def produce(url: str):
res = urlopen(url)
html = res.read()
return html
# Your Code Here
with futures.ThreadPoolExecutor(max_workers=4) as pool:
htmls = pool.map(produce, urls)
#total = sum(len(html) for html in htmls)
total = 0
for html in htmls:
total += len(html)
print(f'Sum of bytes is {total}')
Sum of bytes is 207557 Sum of bytes is 1012144
ProcessPoolExecutor¶
from concurrent import futures
from time import sleep
def task(delay):
sleep(delay)
return 42
if __name__ == '__main__':
with futures.ProcessPoolExecutor(max_workers=1) as pool:
future = pool.submit(task, 5)
while not future.done():
print("Waiting...")
sleep(1)
print(future.result())
from concurrent import futures
from time import sleep
def task(delay):
sleep(delay)
return 42
if __name__ == '__main__':
with futures.ProcessPoolExecutor(max_workers=5) as pool:
future = pool.map(task, [5,5,5,5,5])
print(list(future))
Differences between Threading and Concurrent¶
Level of Abstraction:
threading
: A lower-level module that provides more control but requires more boilerplate code.concurrent.futures
: Provides a higher-level abstraction with easier management of thread and process pools.
Executors:
threading
: Does not provide executors; you manually create and manage threads.concurrent.futures
: ProvidesThreadPoolExecutor
andProcessPoolExecutor
for easier concurrency management.
Task Management:
threading
: Requires explicit thread creation and management.concurrent.futures
: Uses futures to manage and retrieve results of asynchronous tasks.
Use Cases:
threading
: Suitable for simple I/O-bound tasks where low-level control is needed.concurrent.futures
: Ideal for both I/O-bound and CPU-bound tasks where a higher-level interface simplifies the code.
Summary:
- Use
threading
when you need more control over threads and are dealing with I/O-bound tasks. - Use
concurrent.futures
when you prefer a higher-level, easier-to-use interface for managing concurrency and need to handle both I/O-bound and CPU-bound tasks effectively.
Testing¶
pytest Fundamentals¶
! pip install pytest
Requirement already satisfied: pytest in ./.venv/lib/python3.10/site-packages (8.2.2) Requirement already satisfied: iniconfig in ./.venv/lib/python3.10/site-packages (from pytest) (2.0.0) Requirement already satisfied: packaging in ./.venv/lib/python3.10/site-packages (from pytest) (24.1) Requirement already satisfied: pluggy<2.0,>=1.5 in ./.venv/lib/python3.10/site-packages (from pytest) (1.5.0) Requirement already satisfied: exceptiongroup>=1.0.0rc8 in ./.venv/lib/python3.10/site-packages (from pytest) (1.2.1) Requirement already satisfied: tomli>=1 in ./.venv/lib/python3.10/site-packages (from pytest) (2.0.1) [notice] A new release of pip is available: 24.0 -> 24.1 [notice] To update, run: pip install --upgrade pip
SUT: factorial(num)¶
SUT = System Under Test
# %%writefile mathutils.py
def factorial(num):
if not isinstance(num, int):
raise TypeError("Argument must be int")
if num == 0:
return 1
else:
return factorial(num-1) * num
print(factorial(8)) # factorial(3) == 3 * 2 * 1
40320
Basic Tests¶
%%writefile pytest_fundamentals.py
import pytest
from mathutils import factorial # import mathutil file created above
def test_factorial_of_one():
assert factorial(1) == 1
def test_factorial_of_three():
got = factorial(3)
expected = 7 # (!) # 6 is correct
assert expected == got
def test_raises_typeerror_for_invalid_argument():
with pytest.raises(TypeError): # testing raising an exception
factorial(2.5)
Overwriting pytest_fundamentals.py
Launching Tests:¶
! pytest pytest_fundamentals.py
============================= test session starts ============================== platform darwin -- Python 3.10.10, pytest-8.2.2, pluggy-1.5.0 rootdir: /Users/a563420/python_training/advanced_python plugins: anyio-4.4.0 collected 3 items pytest_fundamentals.py ..F [100%] =================================== FAILURES =================================== __________________ test_raises_typeerror_for_invalid_argument __________________ def test_raises_typeerror_for_invalid_argument(): > with pytest.raises(TypeError): # testing raising an exception E Failed: DID NOT RAISE <class 'TypeError'> pytest_fundamentals.py:15: Failed =========================== short test summary info ============================ FAILED pytest_fundamentals.py::test_raises_typeerror_for_invalid_argument - Failed: DID NOT RAISE <class 'TypeError'> ========================= 1 failed, 2 passed in 0.30s ==========================
Exercise¶
Write as many tests as possible for the function below.
def div(a, b):
return a / b
%%writefile pytest_div.py
### Solution
import math
import pytest
def div(a, b):
return a / b
def test_dividing_4_by_2_gives_2():
assert div(4, 2) == 2
def test_dividing_two_integers_gives_mathematically_correct_result():
assert div(3, 2) == 1.5
def test_dividing_two_integers_returns_float_even_when_result_is_integer():
assert isinstance(div(4, 2), float)
def test_raises_error_on_division_by_zero():
with pytest.raises(ZeroDivisionError):
div(2, 0)
def test_dividing_negative_numbers():
assert div(-3, -2) == 1.5
def test_dividing_infinities_gives_nan():
assert math.isnan(div(float('inf'), float('inf')))
def test_dividing_infinity_gives_infinity():
inf = float('inf')
assert div(inf, 2) == inf
def test_dividing_infinity_by_zero_raises_an_error():
inf = float('inf')
with pytest.raises(ZeroDivisionError):
div(inf, 0)
Overwriting pytest_div.py
! pytest pytest_div.py -q
........ [100%] 8 passed in 0.03s
Exercise: @suppress_output¶
suppress_output implementation¶
from contextlib import contextmanager
import sys
@contextmanager
def suppress_output():
stdout = sys.stdout
sys.stdout = None
try:
yield stdout
finally:
sys.stdout = stdout
with contextmanager as context: context # context = yield ... from context manager decorated function
when close intendation, execute finaly¶
Example Usage¶
print("A")
with suppress_output() as s:
print("B")
s.write("asdf\n")
print("C")
A asdf C
import sys
print("A", sys.stdout)
with suppress_output() as s:
print("B")
s.write(f'BB {sys.stdout}\n')
s.write(f's = {s}\n')
print("C", sys.stdout)
A <ipykernel.iostream.OutStream object at 0x102a9f5b0> BB None s = <ipykernel.iostream.OutStream object at 0x102a9f5b0> C <ipykernel.iostream.OutStream object at 0x102a9f5b0>
Excersise: What to test?¶
- sys.stdout should be changed in B.
- If an error is raised in B, suppress_output should not handle it.
- sys.stdout should be restored in C?
- sys.stdout should be restored in C even in case of an error.
- s should be the original stdout.
### Solution
import pytest
from contextlib import contextmanager
import sys
@contextmanager
def suppress_output():
stdout = sys.stdout
sys.stdout = None
try:
yield stdout
finally:
sys.stdout = stdout
def test_stdout_should_change_inside_suppress_output():
before = sys.stdout
with suppress_output():
assert sys.stdout is not before
assert sys.stdout is before
def test_error_is_propagated_outside_of_suppress_output():
before = sys.stdout
with pytest.raises(Exception):
with suppress_output():
raise Exception
assert sys.stdout is before
def test_stdout_should_be_None_inside_suppress_output():
with suppress_output():
assert sys.stdout is None
def test_s_should_originale_inside_suppress_output():
before = sys.stdout
with suppress_output() as s:
assert s is before
def test_suppress_output():
before = sys.stdout
with pytest.raises(Exception):
with suppress_output() as s:
assert sys.stdout is not before
assert sys.stdout is None
assert s is before
raise Exception
assert sys.stdout is before
Miscelanous¶
limitations of file vs file.read()¶
func attribute showing all passed attributes¶
# print(globals().items())
def foo(a, b, *args, **kwargs):
test_Variable = 67
print(locals().items())
foo(1,2, 3, 4)
dict_items([('a', 1), ('b', 2), ('args', (3, 4)), ('kwargs', {}), ('test_Variable', 67)])
a: int = 5
c = (a) # == a and not tuple[a,]
print(c)
print(type(c))
c = (a,)
print(c)
print(type(c))
5 <class 'int'> (5,) <class 'tuple'>
classmethid vs instance metho¶
class Przepis: # class definition (przepis)
waga = 0 # int
def __init__(self):
self.waga = 1000
def instance_method(self): # metoda do wykonania na ciescie (self -> ciasto)
print('instance method')
print(f'waga: {self.waga}')
@classmethod
def class_method(cls): # metoda do wykonania na przepisie (cls -> przepis)
print('class_method')
cls.waga = 50
# Przepis.instance_method() ==> Type Error (nie ma ciasta) , brak instancji klasy
ciasto = Przepis() # class instance (ciasto) -> metoda __init__
ciasto.instance_method()
print(Przepis.waga)
Przepis.class_method()
print(Przepis.waga)
instance method waga: 1000 0 class_method 50
inherit from object example¶
class Foo: # class Foo(object):
pass
class Boo(object):
pass
class Zoo():
pass
f = Foo()
b = Boo()
z = Zoo()