Advanced Python Training at VMWare - Day 2

May 28-30, 2018 Vikrant Patil

These notes are available online at http://notes.pipal.in/2018/vmware-advanced-may

© Pipal Academy LLP

Day 1 | Day 2 | Day 3

Python distribution

We will be using anaconda (python 3) distribution for this training. it can be downloaded from

https://www.anaconda.com/download/

decorators more examples

In [1]:
import requests

def download(url):
    r = requests.get(url)
    return r

Handling Exceptions

In [2]:
try:
    a = int(input("Enter some integer:"))
except ValueError as v:
    print(v)
    print("Integer value expected...")
    

Enter some integer:er
invalid literal for int() with base 10: 'er'
Integer value expected...
In [4]:
try:
    a = int(input("Enter some integer:"))
except ValueError as v:
    print(v)
    print("Integer value expected...")
except RuntimeError:
    pass
except Exception:
    pass
else:
    # executed only if no exception occurs
    pass
finally:
    #will be executed compulsory irrespective of whether exception occurs 
    # or not
    pass

Enter some integer:fg
invalid literal for int() with base 10: 'fg'
Integer value expected...
In [5]:
def raiselist():
    raise []
In [11]:
try:
    raiselist()
except list as l:
    print(l)

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
<ipython-input-11-d6fc87c9164c> in <module>()
      1 try:
----> 2     raiselist()
      3 except list as l:

<ipython-input-5-e0b587bc00bb> in raiselist()
      1 def raiselist():
----> 2     raise []
      3 

TypeError: exceptions must derive from BaseException

During handling of the above exception, another exception occurred:

TypeError                                 Traceback (most recent call last)
<ipython-input-11-d6fc87c9164c> in <module>()
      1 try:
      2     raiselist()
----> 3 except list as l:
      4     print(l)

TypeError: catching classes that do not inherit from BaseException is not allowed
In [ ]:
try:
    raiselist()
except TypeError as t:
    print(t)
In [7]:
l = []
In [8]:
l[1]

---------------------------------------------------------------------------
IndexError                                Traceback (most recent call last)
<ipython-input-8-f32270f6c8bb> in <module>()
----> 1 l[1]

IndexError: list index out of range
In [9]:
try:
    l[1]
except IndexError as e:
    print(e)

list index out of range

problem

  • Implement a decorator with_retries which tries to call function five times. and gives up after that.
from urllib.request import urlopen
@with_retries
def download(url):
    r = urlopen(url)
    return r

download("http://google.com/nosuchurl")
failed -- 1 retrying
failed -- 1 retrying
failed -- 1 retrying
failed -- 1 retrying
failed -- 1 retrying
giving up!
  • Write a decorator timeit which times given function. make use of time module
@timeit
def foo():
    ...
    ...

foo()
Time taken to execute foo = 2.0 seconds
In [12]:
import time
In [13]:
time.time()
Out[13]:
1527569196.1871488
In [14]:
import time
def with_retries(func):
    def wrapper(*args):
        for i in range(5):
            try:
                return func(*args)
            except Exception as e:
                print(e, "Retrying")
                time.sleep(1)
        print("Giving up!")
    
    return wrapper
In [15]:
from urllib.request import urlopen

@with_retries
def download(url):
    return urlopen(url)
In [16]:
download("http://googl.com/nosuchurl.ksjdlkaj")

HTTP Error 404: Not Found Retrying
HTTP Error 404: Not Found Retrying
HTTP Error 404: Not Found Retrying
HTTP Error 404: Not Found Retrying
HTTP Error 404: Not Found Retrying
Giving up!

from urllib.request import urlopen

@with_retries(tries=5, wait=1)
def download(url):
    return urlopen(url)

@vm_running(vmparameters..)
def get_some_info_from_vm():
    pass
In [22]:
def with_retries(tries=5, wait=0.5):
    
    def decor(func):
        
        def wrapper(*args):
            for i in range(tries):
                try:
                    return func(*args)
                except Exception as e:
                    print(e, "retrying...")
                    time.sleep(wait)
            print("Giving up!")
        return wrapper
    
    return decor
In [23]:
from urllib.request import urlopen
@with_retries(tries=2, wait=1)
def download(url):
    return urlopen(url)
In [24]:
download("http://google.com/xkjhfdshfh")

HTTP Error 404: Not Found retrying...
HTTP Error 404: Not Found retrying...
Giving up!
In [25]:
download()

download() missing 1 required positional argument: 'url' retrying...
download() missing 1 required positional argument: 'url' retrying...
Giving up!
In [26]:
import functools
In [27]:
#functools.partial()
In [28]:
def foo(a,b,c):
    pass
In [29]:
foo1 = functools.partial(foo, 2,3)
In [30]:
foo1
Out[30]:
functools.partial(<function foo at 0x7f33a3a40d08>, 2, 3)
In [31]:
foo1(4)
In [32]:
def foo(a,b,c):
    return a+b+c
In [33]:
f = functools.partial(foo, 3, 4)
In [34]:
f(5)
Out[34]:
12

commands ..example

In [36]:
%%file functions.py


def cat(file):
    with open(file) as f:
        print(f.read())
        
        
def head(file, n):
    with open(file) as f:
        for i in range(int(n)):
            print(f.readline(), end="")
            

def grep(pattern, file):
    with open(file) as f:
        for line in f:
            if pattern in line:
                print(line, end="")

Overwriting functions.py
In [45]:
%%file commands.py
import sys

commandcache = {}

def command(func):
    commandcache[func.__name__] = func
    return func

def help_():
    print("Available commands are..")
    for cmd in commandcache:
        f = commandcache[cmd]
        print("{cmd}:{doc}".format(cmd=cmd, doc=f.__doc__))

def main():
    cmd = sys.argv[1]
    if cmd in commandcache:
        commandcache[cmd](*sys.argv[2:])
    else:
        print("Invalid command,",cmd)
        help_()

Overwriting commands.py
In [50]:
%%file functions.py
from  commands import command, main

@command
def cat(file):
    """
    cat filename .. displays file contents 
    """
    with open(file) as f:
        print(f.read())
        
@command        
def head(file, n):
    """
    head filename n .. displays first n lines of file
    """
    with open(file) as f:
        for i in range(int(n)):
            print(f.readline(), end="")
            

def grep(pattern, file):
    with open(file) as f:
        for line in f:
            if pattern in line:
                print(line, end="")


if __name__ == "__main__":
    main()
    

Overwriting functions.py
In [51]:
!python functions.py cat tables.csv

1,2,3,4,5,6,7,8,9,10
2,4,6,8,10,12,14,16,18,20
3,6,9,12,15,18,21,24,27,30
4,8,12,16,20,24,28,32,36,40
5,10,15,20,25,30,35,40,45,50
In [52]:
!python functions.py head tables.csv 2

1,2,3,4,5,6,7,8,9,10
2,4,6,8,10,12,14,16,18,20
In [53]:
!python functions.py grep hello tables.csv

Invalid command, grep
Available commands are..
cat:
    cat filename .. displays file contents 
    
head:
    head filename n .. displays first n lines of file

iterators and generators

Just observe for loop in python

In [54]:
for n in range(5):
    print(n)

0
1
2
3
4
In [55]:
for c in "string":
    print(c)

s
t
r
i
n
g
In [56]:
numbers = range(5)

Iteration protocol

In [57]:
nums = iter(range(5))
In [58]:
type(nums)
Out[58]:
range_iterator
In [59]:
next(nums)
Out[59]:
0
In [60]:
next(nums)
Out[60]:
1
In [61]:
next(nums)
Out[61]:
2
In [62]:
next(nums)
Out[62]:
3
In [63]:
next(nums)
Out[63]:
4
In [64]:
next(nums)

---------------------------------------------------------------------------
StopIteration                             Traceback (most recent call last)
<ipython-input-64-193ae7edbe7d> in <module>()
----> 1 next(nums)

StopIteration:
In [65]:
def digits(n):
    while n >= 0:
        yield n
        n -= 1
In [66]:
def digits(n):
    print("start digits...")
    while n >= 0:
        print("before yield")
        yield n
        print("after yield")
        n -= 1
    print("after while loop")
In [67]:
d = digits(5)
In [68]:
type(d)
Out[68]:
generator
In [69]:
next(d)

start digits...
before yield
Out[69]:
5
In [70]:
next(d)

after yield
before yield
Out[70]:
4
In [71]:
next(d)

after yield
before yield
Out[71]:
3
In [72]:
next(d)

after yield
before yield
Out[72]:
2
In [73]:
next(d)

after yield
before yield
Out[73]:
1
In [74]:
next(d)

after yield
before yield
Out[74]:
0
In [75]:
next(d)

after yield
after while loop

---------------------------------------------------------------------------
StopIteration                             Traceback (most recent call last)
<ipython-input-75-57d0022d53db> in <module>()
----> 1 next(d)

StopIteration:
In [78]:
for d in digits(3):
    print(d)

start digits...
before yield
3
after yield
before yield
2
after yield
before yield
1
after yield
before yield
0
after yield
after while loop
In [79]:
d = digits(2)
In [80]:
next(d)

start digits...
before yield
Out[80]:
2
In [81]:
next(d)

after yield
before yield
Out[81]:
1
In [82]:
next(d)

after yield
before yield
Out[82]:
0
In [83]:
next(d)

after yield
after while loop

---------------------------------------------------------------------------
StopIteration                             Traceback (most recent call last)
<ipython-input-83-57d0022d53db> in <module>()
----> 1 next(d)

StopIteration:
In [84]:
next(d)

---------------------------------------------------------------------------
StopIteration                             Traceback (most recent call last)
<ipython-input-84-57d0022d53db> in <module>()
----> 1 next(d)

StopIteration:
In [85]:
list(digits(2))

start digits...
before yield
after yield
before yield
after yield
before yield
after yield
after while loop
Out[85]:
[2, 1, 0]

problem

  • Write a generator function that generates even numbers less than specified number.
e = evens(10)
for i in e:
    print(i, end=",")
0,2,4,6,8
  • Write a function take that consumes n elements from sequence and returns those as a list.
e = evens(20)
take(e, 5)
[0,2,4,6,8]
In [86]:
def evens(n):
    for i in range(2,n,2):
        yield i

def take(seq, n):
    return [next(seq) for i in range(n)]
In [87]:
take(evens(1000), 5)
Out[87]:
[2, 4, 6, 8, 10]
In [88]:
e = evens(20)
In [89]:
take(e, 5)
Out[89]:
[2, 4, 6, 8, 10]
In [90]:
for i in e:
    print(i, end=",")

12,14,16,18,

generator expressions

In [91]:
e = (i for i in range(1000) if i%2==0)
In [92]:
type(e)
Out[92]:
generator
In [93]:
next(e)
Out[93]:
0
In [94]:
sum(e)
Out[94]:
249500
In [95]:
type(e)
Out[95]:
generator
In [96]:
sum(e)
Out[96]:
0
In [97]:
e = (i for i in range(1000) if i%2==0)
In [98]:
max(e)
Out[98]:
998
In [99]:
def integers():
    n = 0
    while True:
        yield n
        n += 1
In [100]:
naturals = integers()
In [101]:
type(naturals)
Out[101]:
generator
In [102]:
next(naturals)
Out[102]:
0
In [103]:
next(naturals)
Out[103]:
1
In [104]:
sum(take(naturals, 10))
Out[104]:
65

decorator using partila

In [ ]:
coimport time
from functools import partial

def with_retries(f=None, retries=5, delay=0):
    if f == None:
        return partial(with_retries, retries=retries, delay=delay)
    
    def g(*args):
        print("retries = ",retries,"delay=", delay)
        for i in range(retries):
            try:
                return f(*args)
            except Exception as e:
                print(f.__name__, args, "failed:", e)
            time.sleep(delay)
        print("Giving up...")
    
    return g 

@with_retries(retries=2, delay=1)
def download():
    pass

Building data pipeline

With generators and iterators

  • Lazy evaluation
  • Optimized memory (as whole sequence is not in memory)
  • We can build laze data processing pipelines
In [105]:
import os

def find(path):
    for dirpath, dirname, files in os.walk(path):
        for f in files:
            yield os.path.join(dirpath, f)

def take(seq, n):
    return [next(seq) for i in range(n)]
In [108]:
files = find("/home/vikrant/programming/explorations/python")
take(files, 5)
Out[108]:
['/home/vikrant/programming/explorations/python/functions.py~',
 '/home/vikrant/programming/explorations/python/setup.py',
 '/home/vikrant/programming/explorations/python/feedbackemail.py~',
 '/home/vikrant/programming/explorations/python/x.txt',
 '/home/vikrant/programming/explorations/python/coroutines.py']
In [109]:
def grep(pattern, seq):
    for item in seq:
        if pattern in item:
            yield item
In [110]:
files = find(".")
pyfiles = grep(".py", files)
take(pyfiles, 5)
Out[110]:
['./trace.py', './fib.py', './commands.py', './fib1.py', './functions.py']
In [111]:
def readlines(files):
    for f in files:
        with open(f) as fd:
            for line in fd:
                yield line
In [112]:
files = find(".")
pyfiles = grep(".py", files)
lines = readlines(pyfiles)
funcs = grep("def ", lines)
take(funcs, 5)
Out[112]:
['def trace(func):\n',
 '    def log(*args):\n',
 '    def wrapper(*args):\n',
 'def fib(n):\n',
 'def command(func):\n']
In [113]:
def count(seq):
    return sum(1 for i in seq)
In [114]:
files = find(".")
pyfiles = grep(".py", files)
lines = readlines(pyfiles)
funcs = grep("def ", lines)
count(funcs)

---------------------------------------------------------------------------
UnicodeDecodeError                        Traceback (most recent call last)
<ipython-input-114-f53bc36d0dfb> in <module>()
      3 lines = readlines(pyfiles)
      4 funcs = grep("def ", lines)
----> 5 count(funcs)

<ipython-input-113-f277828377b3> in count(seq)
      1 def count(seq):
----> 2     return sum(1 for i in seq)

<ipython-input-113-f277828377b3> in <genexpr>(.0)
      1 def count(seq):
----> 2     return sum(1 for i in seq)

<ipython-input-109-e2eb6988f6bf> in grep(pattern, seq)
      1 def grep(pattern, seq):
----> 2     for item in seq:
      3         if pattern in item:
      4             yield item

<ipython-input-111-a191ed54eca5> in readlines(files)
      2     for f in files:
      3         with open(f) as fd:
----> 4             for line in fd:
      5                 yield line

/home/vikrant/usr/local/anaconda3/lib/python3.6/codecs.py in decode(self, input, final)
    319         # decode input (taking the buffer into account)
    320         data = self.buffer + input
--> 321         (result, consumed) = self._buffer_decode(data, self.errors, final)
    322         # keep undecoded input until the next call
    323         self.buffer = data[consumed:]

UnicodeDecodeError: 'utf-8' codec can't decode byte 0xfc in position 4: invalid start byte
In [115]:
import re

def grep(pattern, seq):
    p = re.compile(pattern)
    for item in seq:
        if p.match(item):
            yield item
In [116]:
p = grep(".*\.py$", ["hello.py", "hello.pyc", "htllo.pyt", "python.exe", "dsadsad.pycds"])
list(p)
Out[116]:
['hello.py']
In [117]:
files = find(".")
pyfiles = grep(".*\.py$", files)
lines = readlines(pyfiles)
funcs = grep("def ", lines)
count(funcs)
Out[117]:
10
In [118]:
files = find(".")
pyfiles = grep(".*\.py$", files)
lines = readlines(pyfiles)
funcs = grep("def ", lines)
In [119]:
funcs
Out[119]:
<generator object grep at 0x7f33a3a67e08>
In [120]:
next(funcs)
Out[120]:
'def trace(func):\n'
In [122]:
import os

def find(path):
    for dirpath, dirname, files in os.walk(path):
        for f in files:
            print("yielding from find")
            yield os.path.join(dirpath, f)

def take(seq, n):
    return [next(seq) for i in range(n)]

import re

def grep(pattern, seq):
    p = re.compile(pattern)
    for item in seq:
        if p.match(item):
            print("yielding from grep")
            yield item

def readlines(files):
    for f in files:
        with open(f) as fd:
            for line in fd:
                print("yielding from readlines")
                yield line
In [135]:
files = find(os.getcwd())
pyfiles = grep(".*\.py$", files)
lines = readlines(pyfiles)
funcs = grep("def ", lines)
In [124]:
next(funcs)

yielding from find
yielding from find
yielding from find
yielding from find
yielding from grep
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from grep
Out[124]:
'def trace(func):\n'
In [125]:
next(funcs)

yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from find
yielding from grep
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from readlines
yielding from grep
Out[125]:
'def fib(n):\n'
In [128]:
def createHugeFile(filename):
    with open(filename, "w") as f:
        for i in range(100000000):
            f.write("Some junk statement!\n")
In [132]:
import os

def find(path):
    for dirpath, dirname, files in os.walk(path):
        for f in files:
            yield os.path.join(dirpath, f)

def take(seq, n):
    return [next(seq) for i in range(n)]

import re

def grep(pattern, seq):
    p = re.compile(pattern)
    for item in seq:
        if p.match(item):
            yield item

def readlines(files):
    for f in files:
        with open(f) as fd:
            for line in fd:
                yield line
In [129]:
createHugeFile("/tmp/data.txt")
In [133]:
lines = readlines(["/tmp/data.txt"])
hello = grep("hello python", lines)
In [134]:
next(hello)
Out[134]:
'hello python\n'

problem

In [144]:
def get_paragraphs(lines):
    para = []
    for line in lines:
        if line.strip()=="":
            if para:
                yield "".join(para)
                para = []
        else:
            para.append(line)
    if para:
        yield "".join(para)
In [141]:
count(get_paragraphs(["A\n","B\n","\n","A\n","\n","\n","A\n"]))
Out[141]:
4
In [142]:
count(get_paragraphs(readlines(["pandp12.txt"])))
Out[142]:
2401

Testing

In [151]:
%%file paragraph.py
import requests

def download(url, filename):
    r = requests.get(url)
    with open(filename, "w") as f:
        f.write(r.text)

def count(seq):
    return sum(1 for i in seq)

def get_paragraphs(lines):
    para = []
    for line in lines:
        if line.strip()=="":
            if para:
                yield "".join(para)
                para = []
        else:
            para.append(line)
    if para:
        yield "".join(para)

if __name__ == "__main__":
    url = "https://ia801405.us.archive.org/12/items/prideandprejudic01342gut/pandp12.txt"
    filename = "pnp.txt"
    #download(url, filename)
    with open(filename) as lines:
        print(count(get_paragraphs(lines)))

Overwriting paragraph.py
In [152]:
!python paragraph.py

2202
In [153]:
%%file test_para.py

import paragraph

def square(x):
    return x*x

def test_square():
    assert square(-1) == 1
    assert square(2) == 4

    

Writing test_para.py
In [155]:
!pytest -v test_para.py

============================= test session starts ==============================
platform linux -- Python 3.6.1, pytest-3.0.7, py-1.4.33, pluggy-0.4.0 -- /home/vikrant/usr/local/anaconda3/bin/python
cachedir: .cache
rootdir: /home/vikrant/trainings/2018/vmware-advanced-may, inifile:
collected 1 items 

test_para.py::test_square PASSED

=========================== 1 passed in 0.28 seconds ===========================
In [156]:
%%file test_para.py

import paragraph

def square(x):
    return x*x

def test_square():
    assert square(-1) == 1
    assert square(2) == 4

    square("s")
    

Overwriting test_para.py
In [157]:
!pytest -v test_para.py

============================= test session starts ==============================
platform linux -- Python 3.6.1, pytest-3.0.7, py-1.4.33, pluggy-0.4.0 -- /home/vikrant/usr/local/anaconda3/bin/python
cachedir: .cache
rootdir: /home/vikrant/trainings/2018/vmware-advanced-may, inifile:
collected 1 items 

test_para.py::test_square FAILED

=================================== FAILURES ===================================
_________________________________ test_square __________________________________

    def test_square():
        assert square(-1) == 1
        assert square(2) == 4
    
>       square("s")

test_para.py:11: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

x = 's'

    def square(x):
>       return x*x
E       TypeError: can't multiply sequence by non-int of type 'str'

test_para.py:5: TypeError
=========================== 1 failed in 0.37 seconds ===========================
In [161]:
%%file test_para.py

import paragraph
import pytest

def square(x):
    return x*x

def test_square():
    assert square(-1) == 1
    assert square(2) == 4

    with pytest.raises (TypeError):
        square("s")
        
    
    with pytest.raises (NameError):
        square("s")

Overwriting test_para.py
In [162]:
!pytest -v test_para.py

============================= test session starts ==============================
platform linux -- Python 3.6.1, pytest-3.0.7, py-1.4.33, pluggy-0.4.0 -- /home/vikrant/usr/local/anaconda3/bin/python
cachedir: .cache
rootdir: /home/vikrant/trainings/2018/vmware-advanced-may, inifile:
collected 1 items 

test_para.py::test_square FAILED

=================================== FAILURES ===================================
_________________________________ test_square __________________________________

    def test_square():
        assert square(-1) == 1
        assert square(2) == 4
    
        with pytest.raises (TypeError):
            square("s")
    
    
        with pytest.raises (NameError):
>           square("s")

test_para.py:17: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

x = 's'

    def square(x):
>       return x*x
E       TypeError: can't multiply sequence by non-int of type 'str'

test_para.py:6: TypeError
=========================== 1 failed in 0.35 seconds ===========================
In [165]:
%%file test_para.py

import pytest
from paragraph import count, get_paragraphs

def test_count():
    assert count([]) == 0
    assert count((i for i in range(100))) == 100
    s = (i for i in range(5))
    assert count(s) == 5
    assert count(s) == 0
    
def test_para():
    p = get_paragraphs([])
    assert count(p) == 0
    assert count(get_paragraphs(["\n"])) == 0
    assert count(get_paragraphs(["A\n"])) == 1
    assert count(get_paragraphs(["A\n","B\n"])) == 1
    assert count(get_paragraphs(["A\n","\n"])) == 1
    assert count(get_paragraphs(["A\n","\n","\n","A\n" ]))== 2
    assert count(get_paragraphs(["A"])) == 1

Overwriting test_para.py
In [166]:
!pytest -v test_para.py

============================= test session starts ==============================
platform linux -- Python 3.6.1, pytest-3.0.7, py-1.4.33, pluggy-0.4.0 -- /home/vikrant/usr/local/anaconda3/bin/python
cachedir: .cache
rootdir: /home/vikrant/trainings/2018/vmware-advanced-may, inifile:
collected 2 items 

test_para.py::test_count PASSED
test_para.py::test_para PASSED

=========================== 2 passed in 0.29 seconds ===========================
In [167]:
%%file weekday.py

import datetime

def now():
    return datetime.datetime.now()

def weekday():
    today = now()
    return today.strftime("%A")

Writing weekday.py
In [168]:
import weekday
In [169]:
weekday.weekday()
Out[169]:
'Tuesday'
In [170]:
%%file weekday.py

import datetime

def now():
    return datetime.datetime.now()

def weekday():
    today = now()
    return today.strftime("%A")

Overwriting weekday.py
In [173]:
%%file test_weekday.py

import weekday
import datetime

def test_weekday(monkeypatch):
    today = (2018, 1, 1)
    
    def fakenow():
        return datetime.datetime(*today)
        
    monkeypatch.setattr(weekday, "now", fakenow)
    
    assert weekday.weekday() == "Monday"

Overwriting test_weekday.py
In [174]:
!pytest -v test_weekday.py

============================= test session starts ==============================
platform linux -- Python 3.6.1, pytest-3.0.7, py-1.4.33, pluggy-0.4.0 -- /home/vikrant/usr/local/anaconda3/bin/python
cachedir: .cache
rootdir: /home/vikrant/trainings/2018/vmware-advanced-may, inifile:
collected 1 items 

test_weekday.py::test_weekday PASSED

=========================== 1 passed in 0.01 seconds ===========================
In [192]:
%%file fixture.py
import os
import pytest
def createfile(path_):
    
    with open(path_, "w") as f:
        f.write("one")
        
@pytest.fixture
def path():
    path_ = os.path.join(os.getcwd(),"testdir")
    if not os.path.exists(path_):
        os.mkdir(path_)
    yield path_
    os.removedirs(path_)
    
def test_createfile(path):
    path_ = os.path.join(path, "test.txt")
    createfile(path_)
    assert os.path.exists(path_)==True
    os.remove(path_)

Overwriting fixture.py
In [193]:
!pytest -v fixture.py

============================= test session starts ==============================
platform linux -- Python 3.6.1, pytest-3.0.7, py-1.4.33, pluggy-0.4.0 -- /home/vikrant/usr/local/anaconda3/bin/python
cachedir: .cache
rootdir: /home/vikrant/trainings/2018/vmware-advanced-may, inifile:
collected 1 items 

fixture.py::test_createfile PASSED

=========================== 1 passed in 0.01 seconds ===========================
In [194]:
%%file mymodule.py

print(__name__)

Writing mymodule.py
In [195]:
!python mymodule.py

__main__
In [196]:
import mymodule

mymodule
In [ ]:
In [ ]: