Jim Cheung

reading notes on Python for DevOps: Learn Ruthlessly Effective Automation

Chapter 1. Python Essentials for DevOps

about format:

>>> '{} comes before {}'.format('first', 'second')
'first comes before second'

>>> '{1} comes after {0}, but {1} comes before {2}'.format('first',
                                                           'second',
                                                           'third')
'second comes after first, but second comes before third'

>>> '''{country} is an island.
... {country} is off of the coast of
... {continent} in the {ocean}'''.format(ocean='Indian Ocean',
...                                      continent='Africa',
...                                      country='Madagascar')
'Madagascar is an island.
Madagascar is off of the coast of
Africa in the Indian Ocean'

>>> values = {'first': 'Bill', 'last': 'Bailey'}
>>> "Won't you come home {first} {last}?".format(**values)
"Won't you come home Bill Bailey?"

>>> text = "|{0:>22}||{0:<22}|"
>>> text.format('O','O')
'|                     O||O                     |'
>>> text = "|{0:<>22}||{0:><22}|"
>>> text.format('O','O')
'|<<<<<<<<<<<<<<<<<<<<<O||O>>>>>>>>>>>>>>>>>>>>>|'

f-string

>>> a = 1
>>> b = 2
>>> f"a is {a}, b is {b}. Adding them results in {a + b}"
'a is 1, b is 2. Adding them results in 3'

>>> count = 43
>>> f"|{count:5d}"
'|   43'

>>> padding = 10
>>> f"|{count:{padding}d}"
'|        43'

function with no return:

>>> def no_return():
...     '''No return defined'''
...     pass
...
>>> result = no_return()
>>> print(result)
None

Chapter 2. Automating Files and the Filesystem

read file

In [13]: with open(file_path, 'r') as open_file:
    ...:     text = open_file.readlines()
    ...:

In [35]: import pathlib
In [36]: path = pathlib.Path("/Users/kbehrman/projects/autoscaler/check_pending.py")
In [37]: path.read_text()

pathlib will overwrite a file if it already exists

In [38]: path = pathlib.Path("/Users/kbehrman/sp.config")
In [39]: path.write_text("LOG:DEBUG")

json

In [10]: import json
In [11]: with open('service-policy.json', 'r') as opened_file:
    ...:     policy = json.load(opened_file)

In [13]: from pprint import pprint
In [14]: pprint(policy)

In [17]: with open('service-policy.json', 'w') as opened_file:
    ...:     policy = json.dump(policy, opened_file)

yaml

$ pip install PyYAML

In [18]: import yaml
In [19]: with open('verify-apache.yml', 'r') as opened_file:
    ...:     verify_apache = yaml.safe_load(opened_file)
In [20]: pprint(verify_apache)   

In [22]: with open('verify-apache.yml', 'w') as opened_file:
    ...:     yaml.dump(verify_apache, opened_file)

csv

In [16]: import csv
In [17]: file_path = '/Users/kbehrman/Downloads/registered_user_count_ytd.csv'
In [18]: with open(file_path, newline='') as csv_file:
    ...:     off_reader = csv.reader(csv_file, delimiter=',')

pandas

In [54]: import pandas as pd
In [55]: df = pd.read_csv('sample-data.csv')
In [56]: type(df)
Out[56]: pandas.core.frame.DataFrame

regexp

In[1]: line = '127.0.0.1 - rj [13/Nov/2019:14:34:30 -0000] "GET HTTP/1.0" 200'

In [2]: re.search(r'(?P<IP>\d+\.\d+\.\d+\.\d+)', line)
Out[2]: <re.Match object; span=(0, 9), match='127.0.0.1'>

In [3]: m = re.search(r'(?P<IP>\d+\.\d+\.\d+\.\d+)', line)

In [4]: m.group('IP')
Out[4]: '127.0.0.1'

Use the finditer method to process the log

In [62]: r = r'(?P<IP>\d+\.\d+\.\d+\.\d+)'
In [63]: r += r'- (?P<User>\w+)'
In [64]: r += r'\[(?P<Time>08/Nov/\d{4}:\d{2}:\d{2}:\d{2} [-+]\d{4})\]'
In [65]: r += r' (?P<Request>"GET .+")'

In [66]: matched = re.finditer(r, access_log)
In [67]: for m in matched:
    ...:     print(m.group('IP'))

Dealing with Large Files

In [46]: def line_reader(file_path):
    ...:     with open(file_path, 'r') as source_file:
    ...:         for line in source_file:
    ...:             yield line

In [47]: reader = line_reader('big-data.txt')

In [48]: with open('big-data-corrected.txt', 'w') as target_file:
    ...:     for line in reader:
    ...:         target_file.write(line)

hash

In [62]: import hashlib

In [63]: secret = "This is the password or document text"

In [64]: bsecret = secret.encode()

In [65]: m = hashlib.md5()

In [66]: m.update(bsecret)

In [67]: m.digest()
Out[67]: b' \xf5\x06\xe6\xfc\x1c\xbe\x86\xddj\x96C\x10\x0f5E'

Encrypting Text

$ pip install cryptography

In [1]: from cryptography.fernet import Fernet
In [2]: key = Fernet.generate_key()
In [3]: key
Out[3]: b'q-fEOs2JIRINDR8toMG7zhQvVhvf5BRPx3mj5Atk5B8='


In [4]: f = Fernet(key)
In [5]: message = b"Secrets go here"
In [6]: encrypted = f.encrypt(message)
In [7]: encrypted
Out[7]: b'gAAAAABdPyg4 ... plhkpVkC8ezOHaOLIA=='

In [1]: f = Fernet(key)
In [2]: f.decrypt(encrypted)
Out[2]: b'Secrets go here'

RSA

In [1]: from cryptography.hazmat.backends import default_backend
In [2]: from cryptography.hazmat.primitives.asymmetric import rsa
In [3]: private_key = rsa.generate_private_key(public_exponent=65537,
                                               key_size=4096,
                                               backend=default_backend())

In [4]: private_key
Out[4]: <cryptography.hazmat.backends.openssl.rsa._RSAPrivateKey at 0x10d377c18>

In [5]: public_key = private_key.public_key
In [6]: public_key = private_key.public_key()
In [7]: public_key
Out[7]: <cryptography.hazmat.backends.openssl.rsa._RSAPublicKey at 0x10da642b0>

In [8]: message = b"More secrets go here"
In [9]: from cryptography.hazmat.primitives.asymmetric import padding
In [11]: from cryptography.hazmat.primitives import hashes
In [12]: encrypted = public_key.encrypt(message,
    ...:    padding.OAEP(mgf=padding.MGF1(algorithm=hashes.SHA256()),
    ...:    algorithm=hashes.SHA256(),
    ...:    label=None))

In [13]: decrypted = private_key.decrypt(encrypted,
    ...:    padding.OAEP(mgf=padding.MGF1(algorithm=hashes.SHA256()),
    ...:    algorithm=hashes.SHA256(),
    ...:    label=None))
In [14]: decrypted
Out[14]: b'More secrets go here'

os.path

In [1]: import os
In [2]: cur_dir = os.getcwd() 1

In [3]: cur_dir
Out[3]: '/Users/kbehrman/Google-Drive/projects/python-devops/samples/chapter4'

In [4]: os.path.split(cur_dir) 2
Out[4]: ('/Users/kbehrman/Google-Drive/projects/python-devops/samples',
         'chapter4')

In [5]: os.path.dirname(cur_dir) 3
Out[5]: '/Users/kbehrman/Google-Drive/projects/python-devops/samples'

In [6]: os.path.basename(cur_dir) 4
Out[6]: 'chapter4'

walk up directory tree

In [7]: while os.path.basename(cur_dir):
   ...:     cur_dir = os.path.dirname(cur_dir)
   ...:     print(cur_dir)

def walk_path(parent_path):
    for parent_path, directories, files in os.walk(parent_path):
        print(f"Checking: {parent_path}")
        for file_name in files:
            file_path = os.path.join(parent_path, file_name)
            last_access = os.path.getatime(file_path)
            size = os.path.getsize(file_path)

Chapter 3. Working with the Command Line

There are two dominant ways to interpret bytes during reading. The first, little endian, interprets each subsequent byte as having higher significance (representing a larger digit). The other, big endian, assumes the first byte has the greatest significance and moves down from there.

In [1]: import sys

In [2]: sys.byteorder
Out[2]: 'little'

environment variables

In [1]: import os

In [5]: os.environ.get('LOGLEVEL') 3

In [6]: os.environ['LOGLEVEL'] = 'DEBUG' 4

In [7]: os.environ.get('LOGLEVEL')
Out[7]: 'DEBUG'

Spawn Processes with the subprocess Module

In [1]: cp = subprocess.run(['ls','-l'],
                            capture_output=True,
                            universal_newlines=True)

In [2]: cp.stdout

To modify the script to run a function automatically only when invoked on the command line, but not during import, put the function invocation into the block after the test:

if __name__ == '__main__':
    say_it()

Using sys.argv

import sys

if __name__ == '__main__':
    print(f"The first argument:  '{sys.argv[0]}'")
    print(f"The second argument: '{sys.argv[1]}'")

Using argparse

import argparse

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Echo your input') 1
    parser.add_argument('message',
                        help='Message to echo')

    parser.add_argument('--twice', '-t',
                        help='Do it twice',
                        action='store_true')

    args = parser.parse_args()

    print(args.message)
    if args.twice:
        print(args.message)

Function Decorators

In [2]: def some_decorator(wrapped_function):
   ...:     def wrapper():
   ...:         print('Do something before calling wrapped function')
   ...:         wrapped_function()
   ...:         print('Do something after calling wrapped function')
   ...:     return wrapper
   ...:

In [3]: def foobat():
   ...:     print('foobat')
   ...:
In [4]: f = some_decorator(foobat)
In [5]: f()
Do something before calling wrapped function
foobat
Do something after calling wrapped function

In [6]: @some_decorator
   ...: def batfoo():
   ...:     print('batfoo')
   ...:

In [7]: batfoo()
Do something before calling wrapped function
batfoo
Do something after calling wrapped function

Using click

$ pip install click

import click

@click.command()
@click.option('--greeting', default='Hiya', help='How do you want to greet?')
@click.option('--name', default='Tammy', help='Who do you want to greet?')
def greet(greeting, name):
    print(f"{greeting} {name}")

if __name__ == '__main__':
    greet()

fire

$ pip install fire

import fire

def greet(greeting='Hiya', name='Tammy'):
    print(f"{greeting} {name}")

def goodbye(goodbye='Bye', name='Tammy'):
    print(f"{goodbye} {name}")

if __name__ == '__main__':
    fire.Fire()

Chapter 4. Useful Linux Utilities

Load Testing with molotov

mport molotov

@molotov.scenario(100)
async def scenario_one(session):
    async with session.get("http://localhost:5000") as resp:
        assert resp.status == 200

Debuggers

import ipdb;ipdb.set_trace()

not much interesting stuffs in the later chapsters ..