Coding: Python заметки

• Python.org – основной сайт. Тут можно скачать актуальный python на windows/linux/mac os, посмотреть документацию
• Гвидо ван Россум – творец языка. По сути аналог Линус Торвальдса для Linux.
• Python – the second best language for everything. И это действительно так, включая новые-модные devops, AI, BigData, autotests. При этом нельзя забывать, что 1) любой язык в конечном счете является лишь инструментом для решения задачи и какой-то инструмент может быть лучше, чем текущий 2) зная один язык, проще изучить новый. Подробнее в статье про программирование.
• Repl.it – онлайн интерпретатор.
• PEP8 – правила по написанию кода, важно соблюдать. Помогут:
  • рекомендуется использовать линтеры, которые проверят ваш код (напр. pylint или flake8 для Python)
  • одноименный чекер. Пример стилистики описан в статье про лучшие практики программирования.
• У Python одна из лучших документаций в мире. 
• Название Python не от змеи, а от комедийного шоу. 
• Python2 оффициально deprecated

/home/user/.local/lib/python2.7/site-packages/paramiko/transport.py:33: CryptographyDeprecationWarning: Python 2 is no longer supported by the Python core team. Support for it is now deprecated in cryptography, and will be removed in the next release.

• Python3 по факту python 3000. В этой версии значительно улучшена архитектура. На сегодняшний день все основные библиотеки поддерживают 3’ю версию.
• (python, vmware) Методы взаимодействия с VMware используя python через API: Pyvmomi (official, supported now), pysphere (not official, deprecated from 2013). Сравнение явно в пользу pyvmomi.
• Python GUI automation – pywinauto/pyautogui – автоматизация GUI.

pywinauto is a set of python modules to automate the Microsoft Windows GUI. At its simplest it allows you to send mouse and keyboard actions to windows dialogs and controls.
PyAutoGUI works on Windows, macOS, and Linux, and runs on Python 2 and 3.

• Python 3.8-3.9 требуют установленного Service Pack 1 на Windows 7
• В отличии от многих других языков отступ (indentation) в python очень важен (обозначает блок кода, как скобки в php/c) и неправильный отступ может приводить к ошибкам.  В этом есть большой плюс – код по умолчанию должен быть «красивым».

• Python – язык с динамической типизацией – переменным не обязательно задавать тип.

• Переменные в python являются ссылками на область памяти (подробнее в описании функции id)

• Интерпретатор Python делает поиск значения переменной в последовательности областей видимостей (scope, namespace) по приоритету до первого совпадения, это называется правило LEGB. Из-за такой приоритезации можно переназначить переменные, уже определенные во встроенных/внешних функциях.
  • L (local) – в локальной функции – locals()
  • E (enclosing) – в функции, включающей нашу локальную функцию
  • G (global) – в общем пространстве скрипта – globals()
  • B (built-in) – во встроенных функциях

>>> str
<class 'str'>
>>> str=[]
>>> str
[]

>>> del str # удаляем переменную str
>>> str
<class 'str'>

Поэтому проверка объявления переменной (иногда нужно в коде) осуществляется поиском строки с названием переменной в необходимом scope.

var1_global = "exist"
if 'var1_global' in globals():
   print("yes")

• В python работает механизм garbage collection. Garbage collection – процесс, который удаляет ненужные элементы. К примеру, чистит память (старые участки) при переназначении переменных – на участки памяти не ссылаются переменные, их можно высвобождать.
• None – по сути нулевой элемент. С точки зрения булевой логики является False, точно так же как пустая строка “”, пустой массив [] или ноль 0.
• Заполненный чем то элемент при этом является True и этим можно пользоваться (главное не думать, что если массив заполнен False, то он будет считаться False):

sm_str = "haha"
if sm_str:
   print("{}".format(sm_str))

# вместо
if sm_str != "":
   print("{}".format(sm_str))

• Проверить что в массиве нет False элементов можно через стандартную функцию in.

if False not in l:
  res = True
else:
  res = False

Обучение

• Базовое обучение на py4e.com от Dr. Charles Russell Severance – создателя популярной книги python for everybody и одноименного курса на своем сайте и coursera

Версии

• CPython – эталонная (классическая) реализация языка. Интерпретатор написан на C.
• Jython – специальная версия Python (судя по дате последнего апдейта в 2015 – заброшенная) для виртуальной машины Java позволяет интерпретатору выполняться на любой системе, поддерживающей Java, при этом классы Java могут непосредственно использоваться из Python и даже быть написанными на Python.
• MicroPython – lightweight реализация для микроконтроллеров с малым потреблением памяти, процессора и питания.  Пример девайса есть на канале Dan Bader (известный популяризатор Python). 

it is compact enough to fit and run within just 256k of code space and 16k of RAM

Frameworks

Фреймворки:

• Flask для простых проектов
• Django для всего остального

Install

CentOS 7

Установка python3.6 (поддерживает String Interpolation в виде format и f-strings) на Ubuntu 14.04.

sudo yum install https://centos7.iuscommunity.org/ius-release.rpm
sudo yum install python36
python3.6 -V
sudo yum install python36u-pip # ставим pip и setuptools
sudo apt install python3-pip
sudo pip3.6 install --upgrade pip # обновляем его
sudo pip3.6 install virtualenv
sudo pip3.6 install requests # ставим пакет requests
sudo pip3.6 install beautifulsoup4
sudo python3 -m pip install virtualenv # альтернатива

CentOS 6

Установка python3.6 на CentOS 6.

sudo yum install https://centos6.iuscommunity.org/ius-release.rpm
sudo yum install python36u
python3.6 -V
== other same as CentOS 7 ==

Ubuntu 14.04

Установка python3.6 на Ubuntu 14.04.

sudo add-apt-repository ppa:deadsnakes/ppa
sudo apt-get update
sudo apt-get install python3.6
sudo apt-get install python3-pip # ставим pip и setuptools
== other same as CentOS 7 ==

После установки в коде (#!/usr/bin/env python3.6) или при запуске (python3.6 <script>) указываем конкретную версию. Так же можно сделать эту версию версией по умолчанию для python3 (не рекомендую т.к. на него полагается shell – стоит ввести фигню в консоли чтобы это увидеть) или даже для “просто” python (еще меньше рекомендую) используя утилиту update-alternatives.

~$ python3 -V
Python 3.4.3
~$ sudo update-alternatives --install /usr/bin/python3 python3 /usr/bin/python3.6 1
update-alternatives: using /usr/bin/python3.6 to provide /usr/bin/python3 (python3) in auto mode
~$ sudo update-alternatives --install /usr/bin/python3 python3 /usr/bin/python3.4 2
update-alternatives: using /usr/bin/python3.4 to provide /usr/bin/python3 (python3) in auto mode
~$ sudo update-alternatives --config python3
There are 2 choices for the alternative python3 (providing /usr/bin/python3).
  Selection    Path                Priority   Status
------------------------------------------------------------
* 0            /usr/bin/python3.4   2         auto mode
  1            /usr/bin/python3.4   2         manual mode
  2            /usr/bin/python3.6   1         manual mode
Press enter to keep the current choice[*], or type selection number: 2
update-alternatives: using /usr/bin/python3.6 to provide /usr/bin/python3 (python3) in manual mode
~$ python3 -V
Python 3.6.8

Shebang возможен с конерктной версией python. Важно использовать env – такая запись универсальнее статического задания интерпретатора т.к. env будет находить python автоматически.

#!/usr/bin/env python3.7

Если при установке shebang выдает ошибку, то нужно очистить скрипт от служебного символа \r или ^M. Причем не обязательно он в конце строки на которую ругается. Оболочка интерпретирует эти символы CR как аргументы.

/usr/bin/env: «python3.7\r»: Нет такого файла или каталога 
No such file or directory

# cat -v test.py | head -n 2
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# cat -v test_old.py | head -n 2
#!/usr/bin/env python3^M
# -*- coding: utf-8 -*-^M

Чаще всего помогает запуск не через shebang, а напрямую указав интерпретатор (он эти символы игнорирует). Но правильнее удалить их.

Пример скрипта для очистки.

with open('test.py', 'rb+') as f:
    content = f.read()
    f.seek(0) # переход в самое начало файла
    f.write(content.replace(b'\r', b''))
    f.truncate()

PIP (установка и использование)

Огромное количество модулей входит в стандартную библиотеку python. Те, которые не входят, обычно можно найти в pip, за редким исключением. 

Установка модулей делается через установку из источника, easy_install или pip. Лучшим вариантом насколько понимаю является pip – это аналог gem в Ruby. Он требует отдельную установку себя like “sudo pip3.6 install requests”.

Установка библиотеки.

sudo pip3 install requests # ставим пакет requests
sudo pip3 install beautifulsoup4

Установка библиотеки из файла whl (требует разрешение конфликтов).

pip3 install pywin32-228-cp37-cp37m-win_amd64.whl
pip3 install cffi-1.14.3-cp37-cp37m-win_amd64.whl
pip3 install pycparser-2.20-py2.py3-none-any.whl

Установка нескольких библиотек на основе списка в файле.

#pip3 install -r requirements.txt
#cat requirements.txt
Pillow

Обновление библиотеки.

sudo pip3 install -U requests

Удаление библиотеки.

sudo pip3 uninstall requests

Просмотр установленных библиотек.

pip3 list
Package    Version   
---------- ----------
certifi    2018.11.29
chardet    3.0.4     
idna       2.8       
pip        18.1      
requests   2.21.0    
setuptools 39.0.1    
urllib3    1.24.1   

Ошибки

Если pip не получается поставить pip с ошибкой “No package python-pip available. Error: Nothing to do”, то нужно поставить репу ius.

sudo yum install https://centos7.iuscommunity.org/ius-release.rpm

Если в качестве результата при установке пакета выдает “Consider using the `–user` option or check the permissions.”, значит нужно запускать из под sudo.

sudo pip3.6 install requests 

Если выдает “sudo: pip3.6: command not found”, то надо каждый раз при использовании pip указывать полный путь или сделать symbolic link.

sudo /usr/local/bin/pip3.6 install requests 

Если выдает timeout, то нужно проверить сеть (iptables, proxy, etc).

  Retrying (Retry(total=0, connect=None, read=None, redirect=None, status=None)) after connection broken by 'ConnectTimeoutError(<pip._vendor.urllib3.connection.VerifiedHTTPSConnection object at 0x7f443823edd8>, 'Connection to pypi.org timed out. (connect timeout=15)')': /simple/pytest/

Если не удается установить библиотеку т.к. она не собрана под нашу систему – можно собрать эту библиотеку. При сборке могут поднадобится системные библиотеки (напр. MS Visual C++).

pip3 install bcrypt-3.2.0-cp36-abi3-win_amd64.whl
bcrypt-3.2.0-cp36-abi3-win_amd64.whl is not a supported wheel on this platform.

python setup.py build
python setup.py install
error: Microsoft Visual C++ 14.0 is required. Get it with "Microsoft Visual C++ Build Tools": https://visualstudio.microsoft.com/downloads/
http://go.microsoft.com/fwlink/?LinkId=691126&fixForIE=.exe. 

Интерпретатор

Просмотр версии, запуск в интерактивном режиме. Другие опции можно посмотреть в man python. 

$ python -V # версия второго
Python 2.7.10

$ python3 -V # версия третьего
Python 3.7.2

$ python3 # запуск в интерактивном режиме
Python 3.7.2 (v3.7.2:9a3ffc0492, Dec 24 2018, 02:44:43)
[Clang 6.0 (clang-600.0.57)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>>

#Простейшая математика в интерактивном режиме
>>> x=6
>>> print(x)
6
>>> x=x*6
>>> print(x)
36
quit() # выход

-u – важная опция, которая может пригодиться, особенно для длительно работающих скриптов, вывод которых нужно сохранять. При ее использовании вывод в файл при редиректе в файл будет происходить без буфферизации, как при выводе в shell. В итоге можно применять tail -f на файл и следить за процессом исполнения. В противном случае вывод будет записан в файл только после завершения скрипта. Тут подробнее и альтернативы.

-u Force stdin, stdout and stderr to be totally unbuffered.
python -u

Reserved words

33 штуки

and       del       from      None      True
as        elif      global    nonlocal  try
assert    else      if        not       while
break     except    import    or        with
class     False     in        pass      yield
continue  finally   is        raise
def       for       lambda    return

Виртуальная среда

Позволяет изолировать разные проекты между собой как в рамках версий python, так и в рамках версий библиотек. Аналог vrf в сетях.

Существует несколько способов создания виртуальной среды python, не говоря даже про docker или виртуализацию уровня VM.

virtualenv

sudo pip3.7 install virtualenv

virtualenvwrapper

virtualenvwrapper – упрощает взаимодействие с виртуальными окружениями.

sudo pip3.7 install virtualenvwrapper

После установки добавляем в .bashrc и перезапускаем bash

export VIRTUALENVWRAPPER_PYTHON=/usr/local/bin/python3.7
export WORKON_HOME=~/venv
. /usr/local/bin/virtualenvwrapper.sh

exec bash

Создание виртуального окружения с python3.7 по умолчанию.

mkvirtualenv --python=/usr/local/bin/python3.7 test_env

Удаление виртуального окружения

rmvirtualenv Test

Переход в виртуальное окружение

workon test_env

Выход из виртуального окружения

 deactivate

Просмотр пакетов в виртуальном окружении

lssitepackages

Подключение/импорт библиотек/модулей/классов/функций

• С помощью import подключаем всю системную библиотеку json
• С помощью import подключаем системные библиотеки json, sys, requests
• С помощью from + import из файла net_query.py (в директории исполнения текущего скрипта) подключаем только функцию model из пользовательской библиотеки net_query.
• С помощью from + import из файла LoadManager.py импортируем класс LoadManager.

import json
import json, sys, requets
from net_query import model
from LoadManager import LoadManager

При этом мы так же могли бы импортировать весь модуль net_query и наоборот – часть (одну) функций модуля json. В любом случае импорта исполняется весь код модуля, просто:

• в случае from + import в пространство имен кода импортируется только объекты (переменные) кода конкретной импортируемой функции, а не весь модуль
• при import всего модуля обращение к объектам идет через модуль (import читает файл с модулем и создает объект этого модуля), а при from + import напрямую к объекту. По этой причине нельзя использовать from + import + * т.к. не будет понятно, откуда появился тот или иной объект в коде.

import statement - A statement that reads a module file and creates a module object.
module object - A value created by an import statement that provides access to the data and code defined in a module.

import sys

sys.argv
sys.version


from sys import argv, version

argv
version


#DO NOT USE from sys import *

В случае длинного названия модуля можно использовать import совместно с as.

import concurrent.futures as cf

Например, код ниже импортирует функцию getcwd из модуля os под псевдонимом work_folder.

from os import getcwd as work_folder

При необходимости импорта из несистемной и нелокальной папки можно расширить пространство поиска библиотек используя sys.path.append:

sys.path.append('/home/user/new/some_dri')
from external_lib import external_lib_function

Methods

Best practices

• Использовать Logging вместо print
• Использовать Str.format вместо конкатенации (по факту лучше f-string, подробнее ниже)

Logging

import logging

debug = True
if debug:
  logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)

log.debug("Doing something!")

import logging

logger = logging.getLogger()
formatter = logging.Formatter('%(asctime)s:%(levelname)s:%(filename)s:%(message)s')
fh = logging.FileHandler("log_fname.log")
fh.setFormatter(formatter)
logger.addHandler(fh)
lh = logging.StreamHandler(sys.stdout)
lh.setFormatter(formatter)
logger.addHandler(lh)
logger.setLevel( logging.INFO )

logger.debug('Debug message.')
logger.info('Info message.')
logger.warning('Warning message.')
logger.error('Error message.')

import logging

logger = logging.getLogger()
logger.setLevel( logging.DEBUG )

# add file handler
fh_formatter = logging.Formatter('%(asctime)s:%(levelname)s:%(filename)s:%(message)s')
fh = logging.FileHandler(log_fname_appmix)
fh.setLevel( logging.DEBUG )
fh.setFormatter(fh_formatter)
logger.addHandler(fh)

# add stdout handler
lh_formatter = logging.Formatter('%(asctime)s\t%(message)s', "%Y-%m-%d %H:%M:%S")
lh = logging.StreamHandler(sys.stdout)
lh.setLevel( logging.INFO )
lh.setFormatter(lh_formatter)
logger.addHandler(lh)

DEBUG:urllib3.connectionpool:Starting new HTTPS connection (1): 192.168.1.1:443
DEBUG:paramiko.transport:=== Key exchange agreements ===
DEBUG:paramiko.transport:Kex: curve25519-sha256@libssh.org
DEBUG:paramiko.transport:HostKey: ssh-ed25519
DEBUG:paramiko.transport:Cipher: aes128-ctr
DEBUG:paramiko.transport:MAC: hmac-sha2-256
DEBUG:paramiko.transport:Compression: none

pip install colorlog

Ошибка “TypeError: ‘RootLogger’ object is not callable” может иметь причиной простая ошибка синтаксиса – нужно обращаться к объекту log через метод, соответствующий logging severity (info, debug, etc), а не напрямую.

TypeError: 'RootLogger' object is not callable

log("Hello")
This is wrong. Correct is
log.info("Hello")

Functions

# The first line of the function definition is called the header; the rest is called the body. The header has to end with a colon and the body has to be indented. The body can contain any number of statements. As you might expect, you have to create a function before you can execute it.

def fctn():
    print("lol")

def some_funct(a,b):
   '''some description of function'''
   return(a+b)

print(some_funct.__doc__)

some description of function

Аргументы (переменные) в функцию могут передаваться:

• позиционно – позиция аргумента при вызове функции соответствует описанию функции на входе. Используя * на входе есть возможность приема на входе функции всех переменных или только части и парсинга их уже в функции (по примеру ARGV из системных). Удобство такого подхода – можно написать сразу универсальную функцию, которая работает как с отдельной строкой, так и с массивом данных т.к. строка по умолчанию преобразуется в массив с одним элементом на входе в функцию.

some_funct(a,b)

>> def some_funct(*all_arg):
>>    print(all_arg)

>> a=1
>> b=3
>> c=5
>> some_funct(a,b,c)

(1, 3, 5)

>> def some_funct(a,*all_arg):
>>    print(a, all_arg)

>> a=1
>> b=3
>> c=5

>> some_funct(a,b,c)

1 (3, 5)

• по ключу – позиция не важна, важен ключ, можно передавать переменные в произвольном порядке, главное, чтобы на входе функции ключи соответствовали передаваемым. При передаче чисел или значений Boolean крайне удобный функционал. При этом можно только часть переменных передавать по ключу, но обязательно, чтобы та часть, которая передается по ключу, была в конце, а позиционные аргументы в начале. Кроме того есть вариант передачи в функцию словаря вида ключ: значение используя ** при передаче переменных. В таком случае функция сама распарсит значения из ключей, как и при стандартном вызове функции. Используя же ** на входе в функцию можно принимать часть или все переменные в виде словаря.

some_funct(a=a,b=b)

some_funct(a,b=b)

ndr_benchmark_test(server='127.0.0.1', iteration_duration=780.0, 
title='ndr-experiments-xf5000', verbose=True, max_iterations=10,
allowed_error=1.0, q_full_resolution=2.0,
latency_pps=0, max_latency=0, lat_tolerance=0,
output=None, yaml_file=None, plugin_file=None,
tunables={}, profile='/opt/trex/v2.82/astf/rus_ent_appmix.py', profile_tunables={},
high_mult=100, low_mult=0,


>> def some_funct(**d):
>> 
>> d = {"pos_ip": "1.1.1.1000", "exception_debug": True}
>> some_funct(**d)
>> some_funct(pos_ip = "1.1.1.1000", exception_debug = True)
>> some_funct("1.1.1.1000", True)
'1.1.1.1000' does not appear to be an IPv4 or IPv6 address
False


>> def some_funct(a,**all_arg):
>> print(a, all_arg)

>> a=1
>> b=3
>> c=5

>> some_funct(a,b=5,c=7)
1 {'b': 5, 'c': 7}

Словарь вида ключ-значение (напр. kwargs) легко распарсить в переменная-значение с использованием setattr.

def get_profile(self, tunables, **kwargs):
  # Assign kwargs
  for key, value in kwargs.items():
    setattr(self, key, value)

print(self.var1)
print(self.var2)

При использовании значений по умолчанию для переменных на входе функции, крайне нежелательно использовать массивы и другие изменяемые элементы (если не нужно именно то, что описано ниже).

def some_funct(a,b,c=[]):
   '''some description of function'''
  c.append(a)
  c.append(b)
  return(c)

a = 2
b = 3
print(some_funct(a,b))
print(some_funct(a,b))
print(some_funct(a,b))

[2, 3]
[2, 3, 2, 3]
[2, 3, 2, 3, 2, 3]

VAR basic (type, ID)

• type(var) # узнаем тип переменной

~$ python3
>>> type("Hello weril!")
<class 'str'>
>>> type(666)
<class 'int'>
>>> type(666.6)
<class 'float'>
>>> type("666.6")
<class 'str'>
>>> type(True)
<class 'bool'>
>>> type(Funct)
<class 'function'>
>>> type(None)
<class 'NoneType'>

• isinstance(25,int) # проверяем принадлежность переменной определенному типу
• str(var) # преобразование в string
• chr(var) # возвращает строку (str) с символом Юникода номер которого равен значению аргумента.
• int(var) # преобразование в integer. Перед преобразованием с помощью метода isdigit можно проверить, что данные являются числом (не содержат, например, символов).

>>> var1="10"
>>> var1.isdigit()
True
>>> var1="10a"
>>> var1.isdigit()
False

• float(var) # преобразование в float (число с плавающей ТОЧКОЙ, не запятой)
  • -str преобразование строки в int, строка должна состоять из цифр
  • -int числа в int. c python3 по факту преобразования во float редко требуются т.к. при операциях с float результатом происходит автоматическая трансляция int во float – при делении двух int результатом является float – в Python3 (не Python2!, там он отбрасывает все после точки – сохраняет int тип для результата). Для сохранения остатка в python2 нужно преобразовать цифры во float.

# python3
>>> a=3/2
>>> a
1.5
>>> type(a)
<class 'float'>

# python2
>>> a=3/2
>>> a
1
>>> type(a)
<type 'int'>

# python2
>>> a=float(3)/float(2)
>>> a
1.5
>>> type(a)
<type 'float'>

• print(globals()) или print(locals()) – вывести все текущие глобальные (определенные в основном коде)/локальные переменные (определенные в функции) и их значения – очень удобно для дебага или сохранения test environment в логах (так и использовал).

• id(var) смотрим адрес переменной в памяти. Чаще всего используется только в учебных целях чтобы показать, что разные переменные ссылаются на одно место памяти при равенстве. Лучше всего демонстрируется на списках т.к. они в python изменяемые, в отличии от, например, чисел. При присвоении другого числа/строки переменной всегда переменная создается заново, ей назначается новый участок памяти, а для списка в памяти по старому адресу меняется значение. Причем какой-то родительской связи между копируемым и скопированным элементом не существует – обе переменных просто ссылаются на один участок памяти, который может изменяться при изменении любой из переменных. По этой причине копирование массивов в общем случае бессмысленно и имеет смысл только при использовании метода copy – при его использовании для новой переменной создается отдельный учаток памяти, но и он имеет подводные камни – не копирует вложенные массивы (оставляет только основной). Для этого обычно используют метод deepcopy.

# integer/string
>>> var1=1000 # нужно использовать числа отличные от -5 до 256 т.к. уже хранятся в памяти (одно число всегда имеет один и тот же адрес в памяти - не создается заново), в отличии от чисел вне диапазона (каждый раз создаются в памяти заново)
>>> var2=var1
>>> id(var1)
4359970544
>>> id(var2)
4359970544

>>> var1=1001
>>> id(var1)
4359970480
>>> id(var2)
4359970544

# array
>>> var1=[1,2]
>>> var2=var1
>>> id(var1)
4360043528
>>> id(var2)
4360043528

>>> var1[0]=3
>>> id(var1)
4360043528
>>> id(var2)
4360043528
>>> var2
[3, 2]
>>> var2.append(555)
>>> var2
[3, 2, 555]
>>> var1
[3, 2, 555]
>>> var3=var2.copy()
>>> id(var2)
4317072392
>>> id(var3)
4317830152
>>> var2.append(666)
>>> var2
[3, 2, 555, 666]
>>> var3
[3, 2, 555]

User Input

var = input("what are you? ") # raw_input Python2
print ("Welcome", var)

DIR

dir(var) – с помощью метода dir можем узнать какие методы возможно применить к объекту. Определяется типом объекта

• к строке строковые методы
• к целому числу – целочисленные

var = "test"
print(dir(var))
...['capitalize', 'casefold', 'center', 'count', 'encode']...

var = 1
print(dir(var))
...['from_bytes', 'imag', 'numerator', 'real', 'to_bytes']...

File (работа с файлами, Files)

Создаем файл, если он отсутствует.

if not os.path.exists(fl):
  os.mknod(fl)

read (open default)

описание первой строки: open возвращает handle (курсор) файла на чтение (не указано ничего кроме названия, по умолчанию чтение), read непосредственно читает файл, rstrip удаляет мусор с конца файла (не весь!), split разделяет строки на отдельные элементы массива (списка). Как альтернатива в теории можно было использовать метод readlines (читает файл и создает массив из каждой строки), но он по практике чаще всего хуже т.к. whitespace символы с конца строки им не удаляются. Более классический подход – делать функцию открытия/итерирования по строкам не в одну строку, из минусов – больше строк, из плюсов – в случае если обхем данных большой мы еще до добавления строки в массив можем ее отбросить и не расходовать оперативную память.

def parse_file_to_list(fname):
  l = open(fname).read().rstrip().split("\n")
  return(l)

# classic read to array/list
l = []
with open("test.txt") as f:
   for line in f:
      l.append(line.rstrip())

#берем login из файла, удаляем из него мусор (\n), кодируем в байт-строку
lgn = open('/home/redkin_p/.duser').read().rstrip().encode('utf-8') 

#берем большой файл, кладем его в массив (каждая строка как элемент]), ищем в каждом элементе определенный паттерн (аналог ruby .select), выдергиваем из подпавших строк последний столбец (последний элемент каждого массива) и помещаем его в новый массив
data_list = open('/home/redkin_p/ORANGE').read().rstrip().split("\n")
data_list = list(x for x in data_list if "DGS-3620" in x)
ip_list = []
for s in data_list:
    ip_list.append(s.split("\t")[-1].replace(" ", ""))
print(ip_list)

write (open with w – write, a – append)

w – перезатирает файл и пишет контент в него

a – добавляет содержимое в существующий файл

# string
def write_to_file(text, fname):
    with open(fname, 'a') as the_file:
        the_file.write(text + "\n")

write_to_file(text, fname)

# array
with open('/home/redkin_p/result.csv', 'a') as the_file:
  for n,e in enumerate(interfaces):
    res = '{};{};{};{}\n'.format(e, broadcast[n], multicast[n], unknowncast[n])
    the_file.write(res)

# bytes
def write_bytes_to_new_file(bytes, fname):
    with open(fname, 'wb') as the_file:
        the_file.write(bytes)

# string with new line and datetime (for logs)
def write_to_file_with_new_line_and_dt(text, fname):
    dt = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") # for log
    with open(fname, 'a') as the_file:
        the_file.write(str(dt) + " " + text + "\n")

write + print

Очень полезная функция, когда нужно вывод из print добавить формированием файла с этим print – просто подменяем print на print_write. Имя файла должно быть в GLOBALS или определенно в классе при ООП (self).

def print_write(line):
  print(line)
  write_to_file(line, global_file)

delete (удаление)

import os

os.remove("/home/redkin_p/result.csv")

copy  (копирование)

from shutil import copyfile

copyfile(src, dst)

seek

Перемещает позицию в файле (при итерации по файлу) в самое начало. Нужно редко, но может пригодится (напр. двойное чтение одного файла).

f.seek(0)

change/edit (изменение)

Изменение содержимого файла, как ни странно, не решаемая несколькими простыми строками кода python задача и самый “простой и надежный” способ по факту самый “тупой и прямолинейный”.

# Read in the file
with open('file.txt', 'r') as file :
filedata = file.read()
# Replace the target string
filedata = filedata.replace('ram', 'abcd')
# Write the file out again
with open('file.txt', 'w') as file:
file.write(filedata)

Полезные методы/функции

Методы по удалению:

import os

def delete_file(fname):
  try:
    os.remove("{}".format(fname))
  except:
    pass

def delete_files_in_dir(somedir):
  res = os.listdir("{}".format(somedir))
  for fname in res:
    fname_full_path = "{}/{}".format(somedir,fname)
    delete_file(fname_full_path)

Можно использовать усовершенствованную версию – опциональный exception при ошибке удаления, удаление по маске.

def delete_file(fname, except_if_fail=False):
  if except_if_fail:
    os.remove("{}".format(fname))
  else:
    try:
      os.remove("{}".format(fname))
    except:
      pass

def delete_files_in_dir(somedir, pattern=""):
  res = os.listdir("{}".format(somedir))
  for fname in res:
    fname_full_path = "{}/{}".format(somedir,fname)
    if pattern in fname:
       delete_file(fname_full_path)

Методы по проверке наличия (isfile или exists)/ненулевому размеру (st_size != 0):

import os

def check_file_existance_boolean(file):
  if os.path.isfile(file) and os.stat(file).st_size != 0:
    return True
  else:
    return False

def check_file_existance(file):
  if os.path.isfile(file) and os.stat(file).st_size != 0:
    pass
  else:
    print("Error: нет файла {} или он имеет нулевой размер".format(file))
    quit()

Расчет MD5 файла

import hashlib

def md5(fname):
    hash_md5 = hashlib.md5()
    with open(fname, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b""):
            hash_md5.update(chunk)
    return hash_md5.hexdigest()

Управление сервисами. Запуск функции.

check_and_change_service_status("inactive", "trex_server")

Управление сервисами. Получение статуса.

def get_service_status(service_name):
serv1_status_raw = subprocess.run("systemctl status {}".format(service_name).split(" "), stdout=subprocess.PIPE, stderr=null)
serv1_status = serv1_status_raw.stdout.decode().split("\n")[2].split()[1].replace("failed", "inactive")
return serv1_status

Управление сервисами. Установка статуса.

def set_service_status(expected_service_status, service_name):
if expected_service_status == "inactive":
expected_service_status = "stop"
elif expected_service_status == "active":
expected_service_status = "start"
else:
write_to_file_with_new_line_and_dt("real service_name {} status is {} expected is {}".format(service_name, real_status, expected_service_status), log_fname)
subprocess.run("systemctl {} {}".format(expected_service_status, service_name).split(" "), stderr=null)

Управление сервисами. Управляющая функция.

def check_and_change_service_status(expected_service_status, service_name):
real_status = get_service_status(service_name)
write_to_file_with_new_line_and_dt("real service_name {} is {} expected is {}".format(service_name, real_status, expected_service_status), log_fname)
if real_status != expected_service_status:
set_service_status(expected_service_status, service_name)
real_status = get_service_status(service_name)
if real_status != expected_service_status:
raise Exception("Can't change {} to expected value {}".format(service_name, expected_service_status), log_fname, global_log_fname)
write_to_file_with_new_line_and_dt("successfully changed {} to expected value {}".format(service_name, expected_service_status), log_fname)

CSV

В целом про csv/excel тут.

Парсинг csv (отброс header, delimiter) с функцией reader.

import csv

with open('file.csv') as f:
  reader = csv.reader(f, delimiter=';')
  headers = next(reader)
  for row in reader:
     print(row)

Можно так же перевернуть массив (чтение с конца).

l = reversed(list(csv.reader(f, delimiter=';')))
for row in l:

Использование библиотеки для некоторых кейсов лучше простого split.

• можно использовать csv.DictReader вместо обычного reader для сопоставления значений в header (главное чтобы он был или в файле или заданным опцией fieldnames=[]) конкретным элементам без своего кода.
• данные ниже будут корректно разбиты по элементам при использовании библиотеки т.к. используемый разделитель “;” находится внутри кавычек.

ip;10.10.10.10;"Узловая 1-ая д.1; чердак"

Записываем в файл список в виде строки с разделителями с функцией writerow для объекта writer. Можно использовать опцию quoting (QUOTE_NONNUMERIC, ALL) для обозначения тех данных, которые должны быть в скобках.

writer = csv.writer(f, delimiter=';', quoting=csv.QUOTE_NONNUMERIC)
for line in data:
  writer.writerow(line)

Так же можно записать большое количество данных скопом, используя функцию writerows. В виде данных должен быть список списков (каждая строка = отдельный список).

writer = csv.writer(f, delimiter=';', quoting=csv.QUOTE_NONNUMERIC)
writer.writerows(data)

Есть и аналогичная reader функция DictWriter, которая позволяет записать словарь (формат аналогичный получаемому из DictReader) с использованием writerow/writerows. При использовании есть особенность – header нужно записывать отдельно (writeheader()), даже если он задан в fieldnames к объекту.

Directory (директории, папки)

Смотрим файлы в директории

import os

for fname in os.listdir( sys.argv[1] ):
   print("{}".format(fname))

Копирование директории.

from distutils.dir_util import copy_tree

copy_tree(from_dir, to_dir)

Удаление директории.

import shutil

shutil.rmtree('/folder_name')

Определение текущей директории. Очень полезно для скриптов, которые оперируют относительными путями.

# current directory
import os

os.path.dirname(os.path.realpath(__file__))

Time: date,  month, hours, seconds

Datetime

import datetime
print(datetime.datetime.now())

Date-time to string

import datetime
datetime.datetime.now().strftime("%Y-%m-%d_%H%M%S") # for file_names
dt = datetime.datetime.now().strftime("%Y-%m-%d_%H%M%S.%f")[:-3] # for file_names with milkiseconds
datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") # for general usage

String date to date object

dt = datetime.datetime.strptime('2020-11-04', "%Y-%m-%d").date()
dt = datetime.datetime.strptime('2022-04-28 15:09:04', "%Y-%m-%d %H:%M:%S")

dt_seconds_before = (datetime.datetime.now() - dt).total_seconds()

Today

print(datetime.date.today())

Day of week (день недели), можно и на русском (см. пример в locale)

>>> from datetime import datetime as date
>>> date.today().weekday()
6
>>> date.today().strftime("%A")
'Sunday'

Datetime + seconds.

scheduled_time = datetime.datetime.now() + datetime.timedelta(seconds=5)
print(datetime.datetime.now())
print(scheduled_time)

Month, можно и на русском (см. пример в locale).

>>> date.today().strftime("%B")
'August'

Определение по году (высокосный/не высокосный) и номеру месяца количества дней.

>>> from calendar import monthrange
>>> monthrange(2011, 2)
(1, 28)
>>> monthrange(2012, 2)
(2, 29)

Разность времени в секундах. Может быть полезно для расчета сколько времени прошло с запуска программы/участка кода. Напр.

1. исполняем/спим до тех пор, пока время со старта не приросло на N секунд. Как только время превышено – делаем что-то другое (напр. прекращаем работу).
2. спим вариативное время, т.к. execution в рамках итерации может происходить за разное время, а наша цель, чтобы каждая итерация занимала примерно одинаковое время

# BASIC EXMPL
>>> import time
>>> start_time=time.time()
>>> end_time=time.time() - start_time
>>> end_time
15.806370258331299
>>> end_time=time.time() - start_time
>>> end_time
24.918872117996216

# USAGE 1 тестируем пока не закончилось время
while (time.time() - test_start_time < test_time):
   print("test code")

# USAGE 2 замедляем итерирование в скрипте на основе времени для каждой итерации.
elapsed_time = time.time() - start_time
need_to_sleep = speed - elapsed_time
time.sleep(need_to_sleep)

Math (математика, статистика)

Отходящую от базовой калькуляции можно делать со стандартной библиотекой math. Далее, по аналогии с re (regexp), можно вызывая модуль объект math применять функции (или извлекать данные, например значение числа pi), определенные в этом модуль-объекте через точку (dot-notation).

import math

math.pi
math.sin(radians)
math.log10(ratio)
math.sqrt(2)

The module object contains the functions and variables defined in the module. To access one of the functions, you have to specify the name of the module and the name of the function, separated by a dot (also known as a period). This format is called dot notation.
dot notation - The syntax for calling a function in another module by specifying the module name followed by a dot (period) and the function name.

Округление к большему (round up) делается с помощью функции ceil.

import math
print(math.ceil(4.2))

Округление к меньшему с помощью функции floor. Но можно просто преобразовать float в int.

import math
print(math.floor(4.2))

Среднее значение (average/mean value) для списка (массива).

import statistics

first_values_avr = statistics.mean(first_values)
last_values_avr = statistics.mean(last_values)

Медиана. Почему часто лучше использовать медиану описано тут (с примером).

Пример расчет в python:
>>> import statistics
>>> l = [25, 17, 23, 18, 24, 23, 16, 85, 50]
>>> statistics.mean(l)
31.22222222222222
>>> statistics.median(l)
23

Наглядно между какими значениями находится среднее (23) и медиана (31):
16
17
18
23
23
24
25 
50
85

Процентиль (numpy).

# PYTHON

>> import numpy as np 
>> a = np.array([1,2,3,4,5]) 
>>> p = np.percentile(a, 50) 
>>> print(p) 3.0

Коэффициент корреляции (correlation coefficient).

import numpy

#a = list(map(int, a)) # example for data from file
#b = list(map(int, b)) # example for data from file
a = [1,4,6]
b = [1,2,3]
print(numpy.corrcoef(a,b)[0][1])

Линия тренда в математическом виде (судя по всему рассчитанная на основе метода наименьших квадратов).

Комментарии

В целом о комментариях и их использовании тут.

Simple comment:

# comment

Random

Библиотека random позволяет генерировать псевдо-случайные числа.

The random module provides functions that generate pseudorandom numbers (which I will simply call "random" from here on).

Функция .random к объекту random генерирует случайное число от 0.1 до 1.

The function random returns a random float between 0.0 and 1.0 (including 0.0 but not 1.0).

import random

random.random()

Функция .randint к объекту random генерирует случайное int число от заданного x до y.

The function randint takes the parameters low and high, and returns an integer between low and high (including both).

random.randint(3, 10)

Функция .choice к объекту random выбирает случайный элемент из массива.

To choose an element from a sequence at random, you can use choice

random.choice([1, 4, 2, 5])
random.choice([5,7,9,11,10,31,32,34])


>>> import random
>>> random.choice([5,7,9,11,10,31,32,34])
7
>>> random.choice([5,7,9,11,10,31,32,34])
31
>>> random.choice([5,7,9,11,10,31,32,34])
7
>>> random.choice([5,7,9,11,10,31,32,34])
10
>>> random.choice([5,7,9,11,10,31,32,34])
7

В random есть возможность использовать random seed для генерации псевдослучайного распредения. При этом, в документации python гарантируется, что несмотря на реальные будущие изменения алгоритмов в стандартной библиотеке random, при использовании одного seed value будет получаться одинаковое псевдорандомное распределение. Примеры использования повторяемого рандома (random seed/repeatable random) – токены MFA, генераторы трафика: tcpreplay-edit, isic, TRex, Ixia IxNetwork.

https://docs.python.org/3/library/random.html 
Most of the random module’s algorithms and seeding functions are subject to change across Python versions, but two aspects are guaranteed not to change:
- If a new seeding method is added, then a backward compatible seeder will be offered.
- The generator’s random() method will continue to produce the same sequence when the compatible seeder is given the same seed.

Кодировка (Coding), utf-8, ascii, encode, decode, ENCODING

Подробнее о таблицах символов ascii/unicode и кодировках utf-8/16/32 в словаре.

Всегда лучше указывать кодировку в header файла. Может помочь от ошибок типа “Non-ASCII character ‘\xc2’ in file” и открывать файлы со скриптами в utf-8 на Windows без преобразований в другие кодировки.

# -*- coding: utf-8 -*-

Для кодирования символов в byte строку используем encode. Можно применять на саму строку или на класс str. Если символы не поддерживаются в таблице символов – будет ошибка. Можно указать разные опции (костыли), что делать с символами, которые не распознаются (ignore, replace, namereplace).

>>> print("привет".encode("utf-8"))
b'\xd0\xbf\xd1\x80\xd0\xb8\xd0\xb2\xd0\xb5\xd1\x82'

>>> print(str.encode("привет","utf-8"))
b'\xd0\xbf\xd1\x80\xd0\xb8\xd0\xb2\xd0\xb5\xd1\x82'

>>> print("привет".encode("ascii"))
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
UnicodeEncodeError: 'ascii' codec can't encode characters in position 0-5: ordinal not in range(128)

>>> print("привет".encode("ascii","ignore"))
b''
>>> print("привет".encode("ascii","replace"))
b'??????'
>>> print("привет".encode("ascii","namereplace"))
b'\\N{CYRILLIC SMALL LETTER PE}\\N{CYRILLIC SMALL LETTER ER}\\N{CYRILLIC SMALL LETTER I}\\N{CYRILLIC SMALL LETTER VE}\\N{CYRILLIC SMALL LETTER IE}\\N{CYRILLIC SMALL LETTER TE}'

Для декодирования символов из byte строки используем decode. При декодировании нужно знать исходную кодировку, иначе (если кодировка кодирования и декодирования не совпадают, например utf-8 – utf-16) могут декодироваться кракозябры или может быть ошибка декодирования (можно так же как в случае с кодированием игнорировать недекодируемые символы). Поэтому желательно указывать кодировку в header файла (см. выше). 

>>> print(b'\xd0\xbf\xd1\x80\xd0\xb8\xd0\xb2\xd0\xb5\xd1\x82'.decode("utf-8"))
привет

>>> print(bytes.decode(b'\xd0\xbf\xd1\x80\xd0\xb8\xd0\xb2\xd0\xb5\xd1\x82',"utf-8"))
привет

>>> print(b'\xd0\xbf\xd1\x80\xd0\xb8\xd0\xb2\xd0\xb5\xd1\x82'.decode("utf-16"))
뿐胑룐닐뗐苑

UnicodeDecodeError: 'utf-8' codec can't decode byte 0xff in position 0: invalid start byte
UnicodeDecodeError: 'ascii' codec can't decode byte 0xd0 in position 0: ordinal not in range(128)

hi_unicode = 'привет'
hi_bytes = hi_unicode.encode('utf-8')
hi_bytes.decode('ascii',"ignore")

По умолчанию кодировка для Linux/MAC – UTF-8, для Windows – зависит от языка системы, в случае русского это cp1251.

Lin
>>> import locale
>>> locale.getpreferredencoding()
'UTF-8'

Win
>>> import locale
>>> locale.getpreferredencoding()
'cp1251'

Строку из UTF-8 преобразовать в CP1251 легко. Напр. может пригодиться тогда, когда программа исполняется с utf-8 кодировкой (под исполняется – имеет комментарии/оперирует с базой/сохраняет вывод/etc с utf-8 символами) но какой-то или весь вывод нужно сохранить в cp1251 – формируем какой-нибудь CSV файл, который будет открываться в Windows.

"привет".decode("utf-8").encode("cp1251",'ignore') # convert from UTF-8 to cp1251

Пример изменения locale для отображения дней недели/месяцов на русском стандартными функциями.

>>> import locale
>>> locale.setlocale(locale.LC_ALL, ('ru_RU', 'UTF-8'))
'ru_RU.UTF-8'
>>> date.today().strftime("%B")
'августа'
>>> date.today().strftime("%A")
'воскресенье'

Большинство библиотек, которые работают с внешними данными (файлы, подключения по сети и к базам данными) позволяют указать кодировку данных непосредственно при вызове. Например, subprocess, file. Каким то библиотекам данные обязательно нужны в виде byte (telnetlib).

test = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8') # декодируем полученные байты в строку 

>>> open('/home/redkin_p/sw')
<_io.TextIOWrapper name='/home/redkin_p/sw' mode='r' encoding='UTF-8'>

Особенность python – байт строку с ascii символами python автоматически декодирует (показывает симолы вместо byte) при отображении, при этом не меняя тип. При этом к byte строке можно применить методы строк типо uppper и это создает еще большую иллюзию, что идет работа с с символами, а не байтами. Если символы не ascii  – python показывает сырые байты, как они есть (as is). В случае перемешки ascii и не-ascii происходит отображение в виде символов того, что удалось конвертировать. 

>>> print("hi".encode("utf-8"))
b'hi'
>>> print("привет".encode("utf-8"))
b'\xd0\xbf\xd1\x80\xd0\xb8\xd0\xb2\xd0\xb5\xd1\x82'

>>> print(type("привет".encode("utf-8")))
<class 'bytes'>
>>> print(type("hi".encode("utf-8")))
<class 'bytes'>

>>> print("hello and привет".encode("utf-8"))
b'hello and \xd0\xbf\xd1\x80\xd0\xb8\xd0\xb2\xd0\xb5\xd1\x82'

Telnet

При использовании telnetlib с python 3+ обязательно кодируем в байт строку данные иначе будет выдавать ошибку TypeError: ‘str’ does not support the buffer interface (python 3.4) или TypeError: a bytes-like object is required, not ‘str’ (python 3.6). Это можно сделать поставив символ “b” перед строкой или применив к нему метод encode (с кодировкой utf-8/ascii или без указания кодировки). Полученные данные в str декодируем в байт строку через decode.

#!/usr/bin/env python3.6
#coding: utf-8

import telnetlib

lgn = open('/home/redkin_p/.duser').read().rstrip().encode('utf-8') #берем login из файла, удаляем из него мусор (\n), кодируем в байт-строку
pswd = open('/home/redkin_p/.dpass').read().rstrip().encode('ascii') #берем password из файла, удаляем из него мусор (\n), кодируем в байт-строку

tn = telnetlib.Telnet("10.10.10.10") #подключаемся к узлу 
tn.read_until("UserName:".encode()) #ждем строку "UserName:"
tn.write(lgn + b"\r") #передаем кодированный в байт-строку login
tn.read_until(b"PassWord:") #ждем строку "PassWord:"
tn.write(pswd + b"\r") #передаем кодированный в байт-строку password
tn.read_until(b"#") #ждем строку "#"
tn.write(b"show switch\r") #передаем необходимый запрос
res = tn.read_until(b"#") #забираем данные, можно так же использовать функцию read_very_eager() на объект telnetlib.
tn.close

print(res.decode('ascii')) # 

SSH

Для ssh используем paramiko.

Функции подключения по ssh и отправке команд в созданный connect.

import paramiko

import time

def ssh_login(ssh_ip, ssh_user, ssh_password):
  client = paramiko.SSHClient()
  client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
  client.connect(ssh_ip, username=ssh_user, password=ssh_password)
  return client.invoke_shell(), client

def push_ssh_command(ssh_conn, command, timeout, out_buf, encoding):
  ssh_conn.send(command)
  time.sleep(timeout)
  command_result = ssh_conn.recv(out_buf).decode(encoding, "ignore")
  print(command_result)
  return command_result

После этого можно делать операции.

Отдельными командами:

push_ssh_command(ssh_conn, 'conf t\n', ssh_timeout, 5000, 'utf-8')

Или набором, объединенным в функции:

def shutdown_if(ssh_conn, ssh_timeout, if):
  push_ssh_command(ssh_conn, 'conf t\n', ssh_timeout, 5000, 'utf-8')
  push_ssh_command(ssh_conn, 'int {}\n'.format(if), ssh_timeout, 5000, 'utf-8')
  push_ssh_command(ssh_conn, 'shutdown', ssh_timeout, 5000, 'utf-8')

Закрытие подключения. В примере ssh_login возвращается объект shell и отдельный объект client, на который нужно применять метод close после завершения работы по ssh (client.close()). Делать это на объект shell неправильно – соединение закрывается не полностью и при накомплении произойдет reject! 

You can have multiple channels in a single SSH connection. Hence closing a single channel, won't close whole SSH connection.
You need to call SSHClient.close to close the physical SSH connection (it takes all channels down with it, if any are still open).
ssh.close()

invoke_shell(term='vt100', width=80, height=24, width_pixels=0, height_pixels=0, environment=None)¶
Start an interactive shell session on the SSH server. A new Channel is opened and connected to a pseudo-terminal using the requested terminal type and size.

SCP

Для начала создаем ssh-подключение, например, используя paramiko.

ssh_conn, ssh_client = ssh_login(ssh_ip, ssh_user, ssh_password, ssh_shell_password)

Для загрузки файла используем scp.put, в качестве транспорта используем созданное ssh-подключение.

def scp_put(ssh_conn, fromdir, todir):
  scp = SCPClient(ssh_conn.get_transport())
  scp.put(fromdir, todir)

Для получения файла по scp проще/лучше всего использовать стандартную функцию get. В интернете есть пример с использованием вида получения файла receive, далее чтения content и записи в файл, но по факту в Windows при таком подходе могут быть проблемы с поврежденным tar архивом (crc правильный, но контент поврежден).

Правильно

def scp_get(ssh_conn, fromdir, todir):
  scp = SCPClient(ssh_conn.get_transport())
  scp.get(fromdir, todir)

Нежелательно

scp = scpclient.Read(ssh.get_transport(), localDir)
content = scp.receive( remoteFile )
if content:
  fd = open( '%s/%s' % (localDir, os.path.split(remoteFile)[1]), 'w' )
  fd.write( content )
  fd.close()

Conditional structures

If/else/elif. В случае использования elif не обязателен else (в отличии от Ruby), а подпадание под несколько elif не означает исполнения кода под каждым – только под первым (как ACL в сетевом мире). 

# one-line
if res == "test": print(x)
if "string" in somevar: print("Hit")

# simple
if res == "test":
    print("true")

# if with else
if res == "test":
    print("true")
else:
    print("false")

# if with else with elif
if res == "test":
    print("true")
elif res == "test2":
    pass # do nothing
elif res == "test3":
    print("false3")
else:
    print("false")

# можно делать конструкции и без else (если не подпадет ни под одно из условий, то не отработает никак)
if res == "test":
    print("true")
elif res == "test2":
    print("false")

Multiline Code, Variable, Comment (многострочный код/переменная комменты)

Code (полезно когда много всего в одной строке)

from some_module import 1, 2, 3, 4, 5, 6, \
                        6, 7, 8, 9, 10,  \
                        11, 12, 13,14

Comment

# - обычный однострочный
''' или """ - multiline

Как правильно разбивать длинные строки кода.

The preferred way of wrapping long lines is by using Python’s implied line continuation inside parentheses, brackets and braces.
Long lines can be broken over multiple lines by wrapping expressions in parentheses. These should be used in preference to using a backslash for line continuation.

# 1
my_sum = (0 + 1 + 2
Break up line
+ 3 + 4 + 5
+ 6 + 7 + 8)

# 2
assert True == True, \
'This is a very long error message that should be put on a separate line'

Variable

var = '''line1
line2
line3'''

Очень удобно multiline использовать для более-менее симпатичного пользовательского вывода в CLI. Пример вывода информаци о VM Azure:

print(f'''Key VMs info:
  name: {vm_info["name"]}
  size: {vm_info["hardwareProfile"]["vmSize"]}
  location: {vm_info["location"]}
  proximityPlacementGroup: {ppg}
  nics: {json.dumps(nics, indent=10)}
''')

Числа (Integer/FLOAT, Hex, dec, bin)

Python even or odd (остаток деления)

number % 2 == 0

Округление python. Вторая переменная определяет количество знаков после запятой. В случае ее отсутствия округляет до чистого integer (чаще всего это и нужно).

>>> round(2.65)
3
>>> round(2.65, 1)
2.6

Long

Тип long в python 2x использовался для длинных целы чисел. В общем случае при сравнении типов нужно учитывать что число может быть и int и long и не делать exception (если только exception именно то, что вы хотели – не получить длинное целое число).

В python 3x отказались от Long вообще.

Системы исчисления

Для перевода между системами исчисления может помимо описанных способов использоваться метод format, подробнее в описании функции.

bin – dec

Из десятичного в двоичный (binary – 0b):

>>> bin(128)
'0b10000000'

Обратно – из двоичного в десятичный (decimal):

>>> int('10000000', 2)
128

Из десятичного в шестнадцатеричный (hex – 0x):

>>> hex(255)
'0xff'

Обратно – из шестнадцатеричного в десятичный (decimal):

>>> int('ff', 16)
255

String

PRINT

Стандартная работа функции print

print (x) # вывод переменной х
print (‘hello world’) # вывод текста hello world (можно как в двойных, так и в одинарных скобках)

Опциональные параметры функции print (аргументы надо передавать как ключевые, не позиционные):

• символ в конце строки end – например, позволяет делать print без new_line
• разделитель sep – по аналогии с AWK
• файл file – по умолчанию std.out, можно сделать std.err, в теории так же можно делать вывод сразу в файл, но лучше не делать
• параметр flush – полезно в циклах (печатать сразу после итерации, к примеру, реализация таймера sleep при print на одном line)

>> print("no_new_line", end='&')
no_new_line&
>> print("new_sep", "test", sep=';')
new_sep;test

for i in range(10):
    print(i, end=" ", flush=True)
    time.sleep(1)

0 1 2 3 4 5 6 7 8 9

replace позволяет сделать замену всех символов (по умолчанию), которые подпадают под первое условие, с регулярными выражениями не работает:

.replace("\x08", "")

В третьем аргументе replace можно указать количество подпадений, которое нужно поменять (чаще всего нужно первое):

.replace("\x08", "", 1)

in позволяет сделать поиск строки в подстроке:

if "string" in somevar: print("Hit")

if "vlan" in "switchport access vlan 666":
   print("Yep")

В print перед строкой можно использовать разные опции, особенно популярно это было раньше, но можно встретить и сейчас. Что-то решается очень легко именно так, например – печатаем фиксированное количество символов в строке, количество пробелов зависит от количества символов строки. По факту эта задача сейчас решается через format и более гибко, в сравнении со старым способом (напр. указание выравнивания). Подробнее в format ниже.

# OLD
>>> print('%5s' % 'x')
x
>>> print('%5s' % 'xxx')
xxx
>>> print('%5s' % 'xxxxx')
xxxxx

# NEW
>>> "{0:>5}".format('xxx')
' xxx'

Простая конкатенация с print через + и ,:

print ("Hello"+name+"master") # конкатенация статичного текста и переменных без пробела
print ("Hello", name, "master") # конкатенация статичного текста и переменных через пробел

Конкатенация через .format, довольно удобно, хотя и в начале кажется что хуже ruby string interpolation

res = '{};{};{};{}\n'.format(e, broadcast[n], multicast[n], unknowncast[n])

Но .format очень функциональный – к примеру, можно:

1. указывать размер столбца по количеству символов в каждом (как в bash printf) или указывать размер строки, дополняя ее пробелами если в исходной строке недостаточно символов,
2. делать выравнивание текста (влево <, вправо >,  вверх ^):
3. указывать название переменных (удобно когда их много и/или они похожего формата данные)
4. исключать дублированные данные с использованием именование или индексов
5. переводить из одной системы исчисления в другую
6. используя * передавать данные не в виде отдельных элементов, а всем массивом

1-2
var = "{:<2} {:20} {:4} {:4}".format(1, "FastEthernet0/1", "down", "down")
print(var)
var = "{:<2} {:20} {:4} {:4}".format(2, "FastEthernet0/10", "down", "down")
print(var)
1 FastEthernet0/1 down down
2 FastEthernet0/10 down down

python3
>>> "{0:>5}".format('xxx')
' xxx'
>>> "{0:^5}".format('xxx')
' xxx '

3
var = "{num:<2} {int:20} {pstat:4} {dstat:4}".format(num=1, pstat="down", dstat="down", int="FastEthernet0/1")
print(var)

1 FastEthernet0/1 down down


4
var = "{num:<2} {int:20} {dstat:4} {dstat:4}".format(num=1, dstat="down", int="FastEthernet0/1")
print(var)

1 FastEthernet0/1 down down


var = "1 FastEthernet0/1 down down"
var = "{0:<2} {2:20} {1:4} {1:4}".format(1, "down", "FastEthernet0/1")

1 FastEthernet0/1 down down


5
var = "{:b}.{:b}.{:b}.{:b}".format(192, 168, 1, 1)
var = "{:08b}.{:08b}.{:08b}.{:08b}".format(192, 168, 1, 1)

11000000.10101000.1.1
11000000.10101000.00000001.00000001


6
temp = "{} {} {}"
values = ["1", "2", "3"]
res = temp.format(*values)

1 2 3

Для конкатенации так же можно использовать f-strings (с python 3.6), он поддерживает максимально простой способ string interpolation – прямо в тексте (как в ruby). Мне этот способ нравится больше всего – он выглядит при большом количестве переменных самым чистаемым:

>>> name = "Petr"
>>> print(f"Hello, {name}!")
Hello, Petr!

Так же для конкатенации можно использовать %. Считаю, устаревший в сравнении с format, f-string метод, но лучше +.

var1 = "Hello"
var2 = "World"
print("% s % s" % (var1, var2))

Можно еще так же множить текст, умножив str на int

>>> first = 'Test '
>>> second = 3
>>> print(first * second)
Test Test Test

Downcase/lowercase, upcase/uppercase. Тут и в аналогичных случаях можно вызывать метод напрямую на переменную, а можно на объект str, передав переменную в метод.

var.lower()
var.upper()

str.lower(var)
str.upper(var)

strip – удаляет все whitespace символы в начале и конце строки. rstrip удаляет только справа (в конце строки), lstrip только слева. В качестве аргумента можно указать символ(ы), которые будут удаляться, например, по умолчанию данные методы не удалят пробел U+0008: BACKSPACE ( rstrip(‘\x08’) ).

out = "[- vlan1 --]"
print(out.strip(" [-]"))

vlan1

Вместо регулярных выражений для извлечения данных зачастую можно и нужно использовать (достаточно использовать): Find, Startswith (ниже).

FIND

Можно найти позицию определенной строки/буквы в заданной строке, а потом извлечь данные используя slice и номер позиции.

s = "hello.my@weril.me"
pos = s.find("@")
print(pos)
pos2 = s.find(".", pos) # второй аргумент в find определяет стартовую позицию, иначе в выборку попал бы индекс 5 вместо 14
print(pos2)
res = s[pos+1:pos2]
print(res)

# test.py
8
14
weril

Startswith

Можно сделать поиск по началу строки не только с помощью regexp:

line = "From weril@me.com"
if line.startswith("From "):
    print("choto")

SPLIT

split – по умолчанию split разделяет строку по whitespace (пробел/etc, причем любое количество whitespace рассматривается как одно целое) на массив (список), но можно конкретно указывать конкретный символ.

line.split()
line.split("\n")

>>> line="some more line              exist"
>>> line.split()
['some', 'more', 'line', 'exist']

В сравнении с regexp split будет работать быстрее/занимать меньше строк кода/будет более понятным по коду:

• для ^ или $ использовать split()[0] с доступом к первому или последнему элементу

x.split()[0]

• для того же, после определенного pattern можно использовать двойной split (например, для получения числа, окруженного двумя уникальными для строки pattern – blabsdfsdfbla;user_id:666;user_class:admin;blablalblsdfsdf)

The double split pattern - Sometimes we split a line one way, and then grab one of the pieces of the line and split that piece again.

x.split(";user_id:")[1].split(";user_class:")[0]

Case insensitive

res = re.sub(r".*(cve[,_-]\d\d\d\d[,_-]\d\d\d\d?\d)", '', x, flags=re.IGNORECASE)

При этом для объекта regexp есть split, который позволит решить те задачи, которые сложнее/невозможно решить обычным split.

REGEXP

• База про regexp с примерами. Там в том числе описаны флаги (напр. DOTALL).
• regexp compile (re.compile)
  • для ускорения сейчас не склишком нужен т.к. по факту в современном python сделано ряд операций, по сути делающих compile и не-compile выражения эквивалентными по скорости
  • используется по привычки и для красивого кода (заранее описываем regexp выражение и к нему применяем методы regexp)
  • все методы эквивалентны, только применяются к компилированному объекту, а не объекту re

regex = re.compile(r'som[Ee] reg[Ee]xp')
m = regex.search(line)

Методы

re.search – смотрим, подпадает есть ли regexp в строке, в результате в базовом сценарии получаем true/false (none). Есть так же почти аналогичная функция re.match, которая делает поиск только в начале строки. По факту при подпадении возвращается объект Match, из которого

• можно получить конкретный match (первое подпадение) или match подпавшей группы используя функцию group (нулевой элемент конкретный match, все последующие относятся к группам); причем в новых версиях python3 (c python 3.6 точно, но не работает в python 3.4) groups использовать не обязательно и можно обратится к match объекту по index для извлечения группы (аналог group: нулевой элемент занят под match, все последующие по номеру подпавшей группы)
• можно получить подпавшие группы () как отдельные элементы списка, используя groups
• можно так же получить index start/stop (span) начала/окончания match строки.

import re

line = "sdfsdf fdsf mm 555 fdfm  ffsdf weril.me mg 77 sdfsdf 66"
res = re.search("weril.me", line)
print(res)
print(res.group())

$ python3 ./test.py
<re.Match object; span=(31, 39), match='weril.me'>
weril.me

# добавляем в regexp группы и делаем print groups
res = re.search("(weril.me).*(66)", line)
print(res.groups())
print(res.group(2))
print(res[2])

~$ python3.6 bin/test.py
<_sre.SRE_Match object; span=(30, 54), match='weril.me mg 77 sdfsdf 66'>
('weril.me', '66')
66
66

re.sub – замена символов с поддержкой regexp. Поддерживает счетчик количества замен (по умолчанию бесконечность) – при указании заменит только столько раз. Так же поддерживает группы – можно из какого то вывода извлечь с помощью regexp matching group и сделать подмену.

import re

re.sub(r".*interface gei-0/1/1/", '', text)
re.sub(r".*interface gei-0/1/1/", '', text, count=10)

re.sub(r"(\d+) (\d+)", r'\1:\2', text)

re.findall – получение данных из match. Причем в результате не только первый/последний match, а все в виде list. Есть так же почти аналогичная функция re.finditer, которая возвращает итератор с объектами Match (в разных случаях разный- list/set list, но в любом случае итератор). Используя группировку () (подробнее в статье про regexp) можно получать только часть данных из match:

import re

line = "sdfsdf fdsf mm 555 fdfm ffsdf mg 77 sdfsdf 66"
res = re.findall("[0-9]+", line)
print(res)

$ python3 ./test.py
['555', '77', '66']

line = "kakayato chush = 0.432423"
res = re.findall("^kakayato chush = [0-9.]+", line)
print(res)

$ python3 ./test.py
['kakayato chush = 0.432423']

line = "kakayato chush = 0.432423"
res = re.findall("^kakayato chush = ([0-9.]+)", line)
print(res)

$ python3 ./test.py
['0.432423']

line = "From toster@weril.me at Sat Jan 13 01:01:01 2019"
res = re.findall("^From .*@([^ ]*)", line)
print(res)

$ python3 ./test.py
['weril.me']

re.split – для объекта regexp есть split, который позволит решить те задачи, которые сложнее/невозможно решить обычным split.

re.split(r' +', text)
re.split(r'[ ,|]', text)

Кодировки

• Строка в python3 имеет по умолчанию кодировку unicode, в отличии от python2.
• В python2 для преобразования в unicode нужно перед строкой ставить u.
• В python3 нужно преобразовывать данные, полученные из сети.

Python 3.7.2
>>> line='line'
>>> type(line)
<class 'str'>
>>> line=u'line'
>>> type(line)
<class 'str'>

Python 2.7.10
>>> line='line'
>>> type(line)
<type 'str'>
>>> line=u'line'
>>> type(line)
<type 'unicode'>

Array/LIST/TUPLE/DICTIONARY/set (массивы, списки, словари, кортежи, множества)

• До python 3.6 словари были неупорядоченны – т.е. последовательность элементов после создания могла меняться. Начиная с python 3.6 они упорядочены. Полагаться на это не стоит т.к. старые версии никуда не пропали и не пропадут в ближайшее время. Если нужно создать последовательный словарь (ordereddict), то можно
  • использовать соответствующую функцию класса Collections при создании массива. Конвертации многомерного массива в ordered нет, но можно это сделать с помощью json.

import collections

# создание нового массива
d1 = collections.OrderedDict()
d1['a'] = 'A'
d1['b'] = 'B'
d1['c'] = 'C'
d1['d'] = 'D'
d1['e'] = 'E'
print(d1)

# конвертация многомерного массива
json_pattern = json.loads(json_pattern, object_pairs_hook=collections.OrderedDict)

• • конвертировать элементы словаря, которые должны быть упорядочены, в список – списки сохраняют последовательность

WAS (not order preservation)
{
"x: {"packets": 881160, "size": 110145},
"y": {"packets": 896550, "size": 35862},
}

NOW (order preserved)
[
{"name":"x", "packets": 881160, "size": 110145},
{"name":"y", "packets": 896550, "size": 35862},
]

Про особенности копирования массивов делай поиск по deepcopy. Если вкратце – копирование простым равно не работает, нужно использовать метод deepcopy.

from copy import deepcopy

total_output = {'results': stats, 'config': config}
deep_copy_total_output = deepcopy(total_output)
return deep_copy_total_output

Длина массива или строки (length)

len(arr)

Количество определенных элементов в массиве

arr = [ 1, 2, 3, 1, 5, 1]
print( arr.count(1) )

3

Максимальный/минимальный элемент (работает для элементов в виде str, так и для int; работает не только для array, но и просто для строки)

max(arr)
min(arr)

Индекс искомого элемента массива

>>> var3
[3, 2, 555, 2]
>>> var3.index(555)
2

Сумма элементов массива (должны быть int/float). Если нет – может помочь метод ниже.

sum(arr)

Разделение массивов на две части используя slice (split in two parts/half).

count = len(values)
first_values = values[:count//2]
last_values = values[count//2:]
print(count, len(first_values), len(last_values))

Конкатенация нескольких массивов (списков, кортежей, множеств). Аналогична возможна разность.

>>> a=[1,2]
>>> b=[3,4]
>>> a+b
[1, 2, 3, 4]

Переворачивание (reverse) массива.

reversed(l)

Поиск элементов по шаблону в списке/массиве и возврат нового списка/массива (аналог ruby .select).

data_list = list(x for x in data_list if "DGS-3620" in x)

Преобразование всех элементов массива в int. Для этого можно использовать генераторы, а можно map.

results = map(int, results)
results = [int(x) for x in results]

Список кортежей или словарь (hash) можно создать из объединения двух списков с помощью функции zip. Для списка кортежей можно zip’ить и большее количество list, не только два.

arr1 = [1,2,3]
arr2 = ["kaka", "maka", "myka"]
print(list(zip(arr1,arr2)))
[(1, 'kaka'), (2, 'maka'), (3, 'myka')]

print(dict(zip(arr1,arr2)))
{1: 'kaka', 2: 'maka', 3: 'myka'}

arr3 = ["l", "k", "m"]
print(list(zip(arr1,arr2,arr3)))
[(1, 'kaka', 'l'), (2, 'maka', 'k'), (3, 'myka', 'm')]

Tuple (кортеж) – по сути тот же самый list, только его нельзя изменять и сортировать. Поэтому он быстрее/занимает меньше памяти.

>>> a1
[1, 3, 5]
>>> t1
(1, 3, 5)
>>> import sys
>>> sys.getsizeof(a1)
88
>>> sys.getsizeof(t1)
72

Используются всегда, когда могут с учетом “недостатков”. Часто используются для временных переменных, для данных, которые хочется иметь в исходном виде (например, данные выгруженные из БД).

Tuples is that they're not mutable and that allows them to be stored more densely than lists. Whatever order you put the tuple in when you create it, it stays in that. You can't append, extend, flip, sort it.

We tend to use tuples when we can.

tup = ("raz", "dvas")
print(tup[0])
>> raz
print(max(tup))
>> raz
print(len(tup))
>> 2
tup[0] = 2
>> Traceback (most recent call last):
  File ".\test.py", line 5, in <module>
    tup[0] = 2
TypeError: 'tuple' object does not support item assignment

Из list можно получить кортеж, используя метод tuple (справедливо и обратное с помощью метода list).

>>> a = [1, 2]
>>> tuple(a)
(1, 2)

Кортеж из одного элемента можно создать поставив запятую после него, иначе интерпретатор будет думать, что это число в скобках. Аналогично и с операциями с tuple – например, добавление элемента.

>>> t1=(1)
>>> t1
1
>>> type(t1)
<class 'int'>

>>> t1=(1,)
>>> t1
(1,)
>>> type(t1)
<class 'tuple'>

>>> t1 = (1,3,5)
>>> t1 + (7,)
(1, 3, 5, 7)

Из dictionary так же можно получить список кортежей, используя метод items (справедливо и обратное с помощью метода list). Подробнее о методе в dictionary. 

>>> a.items()
dict_items([('id', 1), ('name', 'Sesame Street'), ('location', None), ('location2', 'some_loc')])

Кортежи (так же как и списки/множества) поддерживают multiple assignment, что используется при итерациях/выводе с функций.

raz, dvas = ("raz", "dvas")
print(raz)
>> raz
print(dvas)
>> dvas

d = {"id": 1, "name":"Sesame Street"}
for key, value in d.items():
  print(key, value)

print(d.items())
dict_items([('id', 1), ('name', 'Sesame Street')])

Кортежи можно объединять.

>>> t1=(1,3,5)
>>> t2=(2,4,6)
>>> t1+t2
(1, 3, 5, 2, 4, 6)

Создание

Для создания списков/словарей и других итерируемых объектов могут применяться генераторы (list comprehension) – например, цикл for внутри списка. Они являются синтетическим сахаром – т.е. не обязательны для создания программы, но зачастую делают код красивее/быстрее.

>>> var1=["line " + str(i) for i in range(10)]
>>> var1
['line 0', 'line 1', 'line 2', 'line 3', 'line 4', 'line 5', 'line 6', 'line 7', 'line 8', 'line 9']

>>> list(range(6,16))
[6, 7, 8, 9, 10, 11, 12, 13, 14, 15]

>>> ip_correct_list = [ip for ip in ip_list if check_ip(ip)]

array of unique values

Один массив – создаем ряд на основе списка (множество set по умолчанию содержит только уникальные значения), преобразуем обратно в список

>>> a1
[1, 2, 3, 1, 2, 3, 3, 3, 5, 1]
>>> set(a1)
{1, 2, 3, 5}
>>> list(set(a1))
[1, 2, 3, 5]

Объединение массивов (union)

Несколько массивов, через объединение массовой получаем список уникальных значений в обоих массивах (дубли между массивами удаляются). Так же используем множества (set) для этой операции.

def union(a, b):
  return list(set(a) | set(b))
  return list( a.union(b) ) # альтернативный способ (требует a и b в виде множеств)

Разность (Удаление)

Разность массивов (находим значения, которые уникальны в первом массиве относительно второго):

def remove(a, b):
    return list(set(a) - set(b)) 
    return list( a.difference(b) ) # альтернативный способ (требует a и b в виде множеств)

Пересечение

Пересечение массивов (находим значения, которые есть в обоих массивах):

def intersect(a, b):
    return(list(set(a) & set(b)))
    return(list( a.intersection(b) ))  # альтернативный способ (требует a и b в виде множеств)

LOOP (WHILE TRUE)

while True:
   ...

Пример использования: инициализируем подключение по SSH,

• если подключились – выходим из цикла “подключений”,
• если exception (напр. timeout), то ждем какое-то время и пытаемся еще раз. Полезно для того, чтобы не ждать фиксированное время перезагрузки устройства, а подключаться как только устройство готово принять подключение.

while True:
  try:
    ssh = SshControl(params)
    break
  except:
    time.sleep(ssh_timeout) # ожидание между попытками подключения по SSH

Split

Описывал выше.

Join

Преобразует массив в строку. Что интересно, join – строковой метод, применяется на строку, но в качестве аргумента использует массив.

delim = "%"
arg_str = delim.join(sys.argv[1:])

join не обязательно через переменную указывать, можно непосредственно строку.

>>> L
['a', 'b', 'c']

>>> ''.join(L)
'abc'

data_list = "\n".join(L)

Sleep

import time

time.sleep(1)

Sleep функция с print счетчиком countdown в одной строке на основе. Работает и для Python 3 и для Python 2 (2.7).

# python3 ./test.py
[*] sleep to get some data... 10 9 8 7 6 5 4 3 2 1


/test.py
import sys
import time

def sleep(sleep_interval, reason=""):
  sys.stdout.write("[*] sleep{}... ".format(reason))
  for i in range(sleep_interval,0,-1):
    sys.stdout.write(str(i)+' ')
    sys.stdout.flush()
    time.sleep(1)
  print("")

sleep(10, " to get some data")

Slice

Получаем от элемента (строки/массива) до элемента, но не включая. Например: получаем первые пять элементов массива (или букв строки). Применений множество.

arr[beginning_of_string/list:end_of_string/list]
arr[:5] #первые пять элементов массива (с нулевого по четвертый)
>>> c
[1, 2, 3, 4, 5, 6]
>>> c[:5]
[1, 2, 3, 4, 5]

arr[0:5] #первые пять элементов массива
s[-3:] #последние 3 элемента
s[:] #копия списка, часто очень полезно
s[1:] #все элементы кроме первого (полезно при получении аргументов и ARGV для исключения названия скрипта)
s[2:-2] #откидываем первые и последние 2
s[::2] #парные элементы
s[1:4:2] #элементы с первого по четвертый с шагом 2

Iteration/Итерация/итерирование

Итерирование работает с итерируемыми (iterable) объектами. Внутри в базовом сценарии (for до окончания элементов) это процесс последовательных next, next, next на один итерируемый объект до exception (StopIteration).

l = [3,4,7,9]
i = iter(l)
print(next(i))
print(next(i))
print(next(i))
print(next(i))
print(next(i))

3
4
7
9
Traceback (most recent call last):
File "/home/redkin_p/bin/test.py", line 11, in <module>
print(next(i))
StopIteration

Без индекса

for sw in sw_list:
    print(sw)

Итерация с индексом или значением ключа dictionary (по умолчанию при итерации словаря берутся только ключи) на основе:

• enumerate – создает кортеж – номер, значение; можно поменять стартовое значение с 0 на кастомное; enumerate полезно использовать для route-map, EEM скриптов, там, где нужно создавать некую последовательность.
• items – items создает кортеж/tuple вида key, value.

for n,e in enumerate(interfaces):
   print(n)

0 str1
1 str2
2 str3

for n,e in enumerate(interfaces, 1):
   print(n)

1 str1
2 str2
3 str3

# two iteration variables (key and value) for dictionary (default only key)
for key, value in d.items():

С помощью range можно легко итерироваться, создавать последовательности (см. генераторы). При указании range одного аргумента оно используется в качестве stop, а в качестве start используется 0, так же можно указать start value и шаг. Причем функция хранит только эти значения, а промежуточные вычисляются -> независимо от размера диапазона он будет занимать фиксированный объем памяти. Совместно с range удобно использовать функцию in – можно проверить, есть ли значение в диапазоне. Так же из range можно получать значения элементов, как будто это список, рассчитывать длину диапазона с помощью len, делать slice. Чтобы реализовать обратную последовательность нужно на range применить функцию reversed или использовать опцию задания шага (шаг -1).

>>> for i in range(5):
...     print(i)
...
0
1
2
3
4

>>> list(range(3))
[0, 1, 2]

>> r1 = range(10)
>> r2 = range(5,10)
>> r3 = range(4,10,2)
>> print(list(r1))
>> print(list(r2))
>> print(list(r3))
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
[5, 6, 7, 8, 9]
[4, 6, 8]

>> print(10 in range(11))
True

>> print(range(10)[9])
9
>> print(range(10)[0])
0

>> print(len(range(10)))
10
>> print(range(10)[0:5])
range(0, 5)

С помощью range можно создать цифровой словарь (предположим, для bruteforce). Например код ниже создает 8-digit dictionary. В коде сделан костыль для добавления в словарь чисел с leading zero 00000000 – 09999999.

for i in range(100000000):
  if (len(str(i))) == 8:
    if str(i)[0] == "1":
      zero_list = list(str(i))
      zero_list[0] = "0"
      zero_int = str("".join(zero_list))
      print(zero_int)
    print(i)

all – выдает True если все элементы True или элемент пустой – что довольно нелогично т.к. пустой объект по умолчанию False (можно проверить функцией bool).

print(all([True, True, False]))
print(all([True, True, True]))
print(all([]))
print(bool([]))

False
True
True
False

any – выдает True если любой элемент True. При этом нулевой элемент считается False, что доставляет (с учетом вышестоящего).

print(any([True, True, False]))
print(any([True, True, True]))
print(any([]))

True
True
False

append – добавление элемента в конец списка

# Одного
sw_list_with_hostname = []
for sw in sw_list:
    sw_list_with_hostname.append(sw)

# Нескольких (через вложенный массив)
sw_list_with_hostname = []
for sw in sw_list:
    sw_list_with_hostname.append([sw, hostname])

extend – добавление нескольких элементов в конец списка (не вложенного массива в виде одного элемента)

>>> var2
[3, 2, 444, 555, 666]
>>> var2.append(777,888,999)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
TypeError: append() takes exactly one argument (3 given)
>>> var2
[3, 2, 555, 666, [777, 888, 999]]

>>> var2.extend([777,888,999])
>>> var2
[3, 2, 555, 666, [777, 888, 999], 777, 888, 999]

insert – добавление элементов на определенную позицию (не в конец)

>>> var2
[3, 2, 555, 666]
>>> var2.insert(2,444)
>>> var2
[3, 2, 444, 555, 666]

pop – удаление последнего элемента или произвольного элемента по id

>>> var2
[3, 2, 555, 666, [777, 888, 999], 777, 888, 999]
>>> var2.pop()
999
>>> var2
[3, 2, 555, 666, [777, 888, 999], 777, 888]
>>> var2.pop(3)
666
>>> var2
[3, 2, 555, [777, 888, 999], 777, 888]

remove – удаление элемента по значению (удаляется первый подпавший элемент)

>>> var2
[3, 2, 555, [777, 888, 999], 777, 888]
>>> var2.remove(777)
>>> var2
[3, 2, 555, [777, 888, 999], 888]

filter – фильтрация массива в новый урезанный массив по заданному условию. К примеру, ищем строки, которые можно интерпретировать как число.

>>> a = ["a", "b", "1", "6"]
>>> list(filter(str.isdigit, a))

['1', '6']

Поиск элементов

Самый простой способ:

7 in a # 7 in array?

Можно так же в dictionary (hash/assoc array) по ключу, что зачастую удобно:

if name in names:
  names[name] = names[name] + 1
else:
  names[name] = 1

Пример

#берем большой файл, кладем его в массив (каждая строка как элемент]), ищем в каждом элементе определенный паттерн, выдергиваем из подпавших строк последний столбец (последний элемент каждого массива) и помещаем его в новый массив
data_list = open('/home/redkin_p/ORANGE').read().rstrip().split("\n")
data_list = list(x for x in data_list if "DGS-3620" in x)
ip_list = []
for s in data_list:
     ip_list.append(s.split("\t")[-1].replace(" ", ""))
print(ip_list)

collections

Аналог sort | uniq –c (количество уникальных объектов)

from collections import Counter
input = ['a', 'a', 'b', 'b', 'b']
c = Counter( input )

sorted, sort

sorted – сортировка list (array), более универсальная в сравнении с sort альтернатива, работает не только с list, а с любым итерируемым (iterable) объектом. Не изменяет исходный list, в отличии, опять же, от sort. Кроме того sorted можно использовать на многомерный массив. Сортировка по умолчанию будет по первому элементу в каждом массиве, но этим можно управлять, например, с помощью использования ключевого аргумента key + itemgetter или с помощью key + lambda.

sort – сортировка list (array), работает только с list, изменяет исходный list

flist.sort()
sorted(flist)

>>> var1
[3, 2, 444, 555, 666, 777, 888, 999]
>>> sorted(var1)
[2, 3, 444, 555, 666, 777, 888, 999]
>>> var1.sort()
>>> var1
[2, 3, 444, 555, 666, 777, 888, 999]

arr = [ [1, 22, 55],
[7, 11, 34],
[2, 111, 23],
[31, 11, 24] ]
print(sorted(arr))

[[1, 22, 55], [2, 111, 23], [7, 11, 34], [31, 11, 24]]

import operator

<same_array>
print(sorted(arr, key=operator.itemgetter(1)))

[[7, 11, 34], [31, 11, 24], [1, 22, 55], [2, 111, 23]]

>>> print(sorted(arr, key=lambda x: x[1]))

[[7, 11, 34], [31, 11, 24], [1, 22, 55], [2, 111, 23]]

через равно sort делать не надо, будет ошибка; sorted можно

flist = flist.sort()
TypeError: 'NoneType' object is not iterable

flist = sorted(flist)

reverse с помощью sort (безусловно можно и с sorted)

flist.sort(reverse=True)
sorted(flist, reverse=True)

Сортировка dictionary (hash, assoc array)

по значению (аналогичное можно и с помощью itemgetter)

names = {}
sorted_x = sorted(names.items(), key=lambda kv: kv[1])

по ключу (аналогичное можно и с помощью itemgetter)

names = {}
sorted_x = sorted(names.items(), key=lambda kv: kv[0])

Чуть оффтопика про lambda – это по сути однострочный эквивалент функции (типа генераторов).

def sum(a,b):
    return a+b

lambda a, b: a+b # equal

Python dictionary (key-value store: hash, associative array, property bag)

Python dictionary – наиболее “powerful” memory collection в Python. По сути это key-value store, который реализован в большинстве ЯП (примеры в header). Такое хранилище позволяет создавать простую in-memory “database” в одной переменной.

Создание

a = {"id": 1, "name":"Sesame Street"}
a = dict(id=1, name="Sesame Street") # ключ должен быть str

# через преобразование
list(str)
list(tuple)
list(set)
list(dictionary)

# на основе массива, не использовать если значение будет arr - иначе arr будет единый для всех ключей (не уникален т.к. ссылка на один участок памяти) - подробнее в примере. В таком случае лучше использовать генератор (по аналоги с list).
>>> arr = [1,2,3]
>>> dict.fromkeys(arr)
{1: None, 2: None, 3: None}
>>> dict.fromkeys(arr, "some_data")
{1: 'some_data', 2: 'some_data', 3: 'some_data'}

# пример почему не стоит использовать массив в качестве value при создании с помощью fromkeys
>>> dct = dict.fromkeys(arr, "some_data")
>>> dct
{1: 'some_data', 2: 'some_data', 3: 'some_data'}
>>> dct[1] = "some_new_data"
>>> dct
{1: 'some_new_data', 2: 'some_data', 3: 'some_data'}
>>> 
>>> dct = dict.fromkeys(arr, [1])
>>> dct
{1: [1], 2: [1], 3: [1]}
>>> dct[1].append(555)
>>> dct
{1: [1, 555], 2: [1, 555], 3: [1, 555]}

# Ключем dictionary может быть только неизменяемая переменная, содержимое которой не изменит свой hash т.к. все операции "внутри" по ключу происходят именно по hash. Это можно увидеть, попытавшись сделать ключ в виде list.
>>> a = {[1,2]: 1, "name":"Sesame Street"}
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
TypeError: unhashable type: 'list'

Добавление элемента

a["id"] = 1

Обновление элемента (или добавление нового, если такого элемента нет; append for dict)

a.update({'some_key': 555})

Удаление элемента

>>> a = {"id": 1, "name":"Sesame Street"}
>>> a
{'id': 1, 'name': 'Sesame Street'}
>>> del a["id"]
>>> a
{'name': 'Sesame Street'}

Получение значения элемента по ключу.

>>> a["name"]
'Sesame Street'

При отсутствии ключа и потенциальной возможности запроса по несуществующему ключу нужно или использовать try except или использовать метод get (возвращает None при отсутствии элемента или заданное во второй переменной к методу значение), в противном случае будет exception. Еще одной безусловной альтернативой (но хуже get – он по сути делает тоже самое без if) является использование in совместно с if – с помощью in мы сможем посмотреть есть ли ключ в словаре.

The pattern of checking to see if a key is already in a dictionary and assuming a default value if the key is not there so common that there is a method called get() that does this for us.

>>> a["location"]
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
KeyError: 'location'
>>> a.get("location")
>>> a.get("location", "some_val")
'some_val'

Очень удобно функцию get использовать при инициализации dictionary, со счетчиком срабатываний элемента. Не нужно расссматривать специальный случай обращения по ключу к отсутствующему элементу (как описано выше).

# simple histogram without if-else
counts = dict()
items = [1,2,2,3,1,1]
for item in items:
  counts[item] = counts.get(item, 0) + 1
print(counts)

~$ test.py
{1: 3, 2: 2, 3: 1}

Так же в таком случае возможно использование функции setdefault. В случае наличия ключа она возвращает его значение, при отсутствии ключа создает его с значением по умолчанию None. В аргументах None можно изменить на что-то конкретное.

>>> a.setdefault("location")
>>> a
{'id': 1, 'name': 'Sesame Street', 'location': None}
>>> a.setdefault("name")
'Sesame Street'
>>> a
{'id': 1, 'name': 'Sesame Street', 'location': None}
>>> a.setdefault("location2", "some_loc")
'some_loc'
>>> a
{'id': 1, 'name': 'Sesame Street', 'location': None, 'location2': 'some_loc'}

Методами keys, values, items можно получить/преобразовать данные исходного словаря в списки. При использовании этих методов нужно учитывать, что они не создают копию данных и при изменении исходного словаря меняются в соответствии с ним. Для защиты от этого результат метода можно преобразовывать в list или работать с копией массива (использовать deepcopy).

>>> a
{'id': 1, 'name': 'Sesame Street', 'location': None, 'location2': 'some_loc'}
>>> a.keys()
dict_keys(['id', 'name', 'location', 'location2'])
>>> a["new_id"] = 2
>>> a.keys()
dict_keys(['id', 'name', 'location', 'location2', 'new_id'])

>>> b = list(a.keys())
>>> a["new_new_id"] = 2
>>> list(a.keys())
['id', 'name', 'location', 'location2', 'new_id', 'new_new_id']
>>> b
['id', 'name', 'location', 'location2', 'new_id']

keys – получить все ключи.

>>> a
{'id': 1, 'name': 'Sesame Street', 'location': None, 'location2': 'some_loc'}

>>> a.keys()
dict_keys(['id', 'name', 'location', 'location2'])

values – получить все значения.

>>> a.values()
dict_values([1, 'Sesame Street', None, 'some_loc'])

items – преобразовать словарь в список кортежей.

>>> a.items()
dict_items([('id', 1), ('name', 'Sesame Street'), ('location', None), ('location2', 'some_loc')])

update – 1) обновляем значения ключей при наличии ключей в существующем массиве 2) создаем пары ключ-значение при отсутствии ключа в существующем массиве

>>> a = {'id': 1, 'name': 'Sesame Street', 'location': None}
>>> a
{'id': 1, 'name': 'Sesame Street', 'location': None}
>>> b = {'id':2, 'password':123456}
>>> a.update(b)
>>> a
{'id': 2, 'name': 'Sesame Street', 'location': None, 'password': 123456}

Python set (множества)

Содержит уникальные значения. Часто используется для задач слияния/разности массивов.

>>> set('tetst')
{'e', 's', 't'}
>>> set([1,2,"ha",2,3,1])
{1, 2, 3, 'ha'} 

Для множеств есть аналогичные list методы только с другими названиями.

>>> s1 = {1,3,5,7}
>>> type(s1)
<class 'set'>
>>> s1.add(9)
>>> s1
{1, 3, 5, 7, 9}
>>> s1.discard(3)
>>> s1
{1, 5, 7, 9}
>>> s1.discard(3) # нет exception в случае отсутствия значения в множестве

Как и в случае с кортежами/словарями из списка/кортежа/словаря можно получить множество и наоборот. Это свойство может использоваться для чистки дублей в списке – преобразовываем его во множество, а потом обратно в список.

>>> s1
{1, 5, 7, 9}
>>> list(s1)
[1, 5, 7, 9]

>>> a1
[1, 2, 3]
>>> set(a1)
{1, 2, 3}

>>> a1
[1, 2, 3, 1, 2, 3, 3, 3, 5, 1]
>>> set(a1)
{1, 2, 3, 5}
>>> list(set(a1))
[1, 2, 3, 5]

Нельзя обращаться по индексу к set т.к. в set не сохраняется позиция элемента. Обходим через list(set).

res_set = set(res_arr)
print(res_set[0])
TypeError: 'set' object is not subscriptable

uniq_res_arr = list(set(res_arr))
print(uniq_res_arr[0])

Thread

Многопоточность можно запустить очень просто

from multiprocessing.dummy import Pool as ThreadPool 

def query_data(ip):
    print(ip)

pool = ThreadPool(10) # количество потоков
pool.map(query_data, sw_list) # запускаем функцию query_data и передаем ей на входе массив sw_list

Если аргументов в функцию нужно передать несколько, можно создать двумерный массив и далее:

1) передать его starmap вместо одномерного и наслаждаться стандартной функцией (starmap требует python версии как минимум 3.3 иначе будет ошибка AttributeError: ‘ThreadPool’ object has no attribute ‘starmap’)

from multiprocessing.dummy import Pool as ThreadPool 

def query_data(ip, hostname):
     print(ip, hostname)

pool = ThreadPool(10) # количество потоков
sw_list_with_hostname = [] # создаем новый массив вида [[ip, hostname], [ip2, hostname]]
pool.starmap(query_data, sw_list_with_hostname) # запускаем функцию query_data и передаем ей на входе массив sw_list_with_hostname

2) передать его pool.map вместо одномерного и разбирать на входе функции (менее удобно)

from multiprocessing.dummy import Pool as ThreadPool 

def query_data(data):
    ip = data[0]
    hostname = data[1]
    print(ip)

pool = ThreadPool(10) # количество потоков
sw_list_with_hostname = [] # создаем новый массив вида [[ip, hostname], [ip2, hostname]]
for sw in sw_list:
sw_list_with_hostname.append([sw, hostname])
pool.map(query_data, sw_list_with_hostname) # запускаем функцию query_data и передаем ей на входе массив sw_list_with_hostname

ARGV

Получаем данные из системных переменных в виде array – путь к исполняемому файлу (всегда в argv[0]) и далее опционально (что передали) – имена файлов, адреса (IP, mail, url), все что угодно.

import sys

print(sys.argv)
print(sys.argv[1])
quit()

Определяем название исполняемого файла без использования argv и полного пути к файлу.

print(os.path.basename(__file__))

Для этой задачи можно так же использовать argparse (есть и другие альтернативы) – позволит сделать аналогию linux утилитам.

python ping_function.py -a 8.8.8.8 -c 5

$ python ping_function.py
usage: ping_function.py [-h] -a IP [-c COUNT]
ping_function.py: error: the following arguments are required: -a

$ python ping_function.py -h
usage: ping_function.py [-h] -a IP [-c COUNT]

Ping script

optional arguments:
  -h, --help  show this help message and exit
  -a IP
  -c COUNT

PSUTIL, OS

Работа с системными процессами из python.

Получить все процессы.

pip3 install psutil
import psutil

psutil.pids()

Проверка на то, что процесс существует в системе.

import os

def check_pid(pid): 
  """ Check For the existence of a unix pid. """
  try:
    os.kill(pid, 0)
    return True
  except OSError:
    return False

Получить cmdline Linux процесса.

def get_pid_cmdline(pid):
  """ Get unix pid cmdline. """
  try:
    cmdline = open(f"/proc/{pid}/cmdline").read()
    return cmdline
  except OSError:
    return ""

Смотрим сессии в текущем VRF. Пример скрипта для вывода информации в формате “netstat/ss”.

psutil.net_connections

# default format
[sconn(fd=3, family=<AddressFamily.AF_INET: 2>, type=<SocketKind.SOCK_STREAM: 1>, laddr=addr(ip='0.0.0.0', port=80), r
addr=(), status='LISTEN', pid=1484)]

# netstat script

import socket
from socket import AF_INET, SOCK_STREAM, SOCK_DGRAM

import psutil

AD = "-"
AF_INET6 = getattr(socket, 'AF_INET6', object())
proto_map = {
(AF_INET, SOCK_STREAM): 'tcp',
(AF_INET6, SOCK_STREAM): 'tcp6',
(AF_INET, SOCK_DGRAM): 'udp',
(AF_INET6, SOCK_DGRAM): 'udp6',
}

def main():
templ = "%-5s %-30s %-30s %-13s %-6s %s"
print(templ % (
"Proto", "Local address", "Remote address", "Status", "PID",
"Program name"))
proc_names = {}
for p in psutil.process_iter(['pid', 'name']):
proc_names[p.info['pid']] = p.info['name']
for c in psutil.net_connections(kind='inet'):
laddr = "%s:%s" % (c.laddr)
raddr = ""
if c.raddr:
raddr = "%s:%s" % (c.raddr)
name = proc_names.get(c.pid, '?') or ''
print(templ % (
proto_map[(c.family, c.type)],
laddr,
raddr or AD,
c.status,
c.pid or AD,
name[:15],
))


if __name__ == '__main__':
main()

Exception

offtop: зачастую вместо exception handling лучше использовать guardian pattern – перед опасным участком кода делать необходимые для корректного исполнения кода проверки. Их можно делать в отдельных if блоках или составляя логическое выражение, где первыми элементами проверяются guardian, а в последующем необходимая логика (последовательность важна, иначе guardian не сработает и будет exception). Подробнее о guardian pattern и short-circuit evaluation можно прочитать тут.

# Guardian
if len(a) < 0:
    quit()
if a[0] == "Added":
    print(a)

# Guardian in compound statement
if len(a) > 0 and a[0] == "Added":
    print(a)

Exception raise (вызов)

Raise можно использовать для генерации exception.

if os.path.exists(workDir):
    raise Exception("Error: Directory already exist.")

exception Handling (обработка)

Отработка на KeyBoardInterrupt exception, далее на ZeroDivisionError. Ниже так же можно было бы добавить блоки else (исполнение в случае если exception не сработал), finally (исполнение в любом случае – сработал exception или нет).

try:
   print("here is untrusted code")
except (KeyboardInterrupt, SystemExit):
   print("KeyBoard exception")
except ZeroDivisionError:
   print("Do not divide by zero")

Отработка на все ошибки (так не рекомендуется делать), без сбора информации о ошибке

try:
    print("here is untrusted code")
except: # catch *all* exceptions
    print("exception")

Отработка на все ошибки, со сбором информации о ошибке (что за ошибка) и traceback (что еще более важно – содержит, напр. указание конкретной строки с ошибкой).

import traceback

try:
    print("here is untrusted code")
except Exception as e:
    tb = traceback.format_exc()
    print(f"exception {e}")
    print(f"traceback {tb}")

Повтор операции N раз в случае exception легко делается через связку for + break.

How to retry after exception?
https://stackoverflow.com/questions/2083987/how-to-retry-after-exception
Do a while True inside your for loop, put your try code inside, and break from that while loop only when your code succeeds.

for i in range(1,exception_retries_count+1):
try:
  <some_not_stable_process>
  break
except Exception as e:
  time.sleep(10)
  write_to_file(f"exception {} was raised (retry ({}/{})".format(e, i, exception_retries_count), log_fname)
  if i == exception_retries_count:
     quit_with_log("Error: maximum retries count reached.", log_fname)

Retiries до success делается с помощью while True + rescue + break:

       while True:
           try:
             <CODE>
           except Exception as e:
             sleep_fun(10, " caught exception ({}), trying again".format(e))
             continue
           break

В функции с exception можно заложить режим debug (писать exception) по передаваемой опции.

def check_ip(pos_ip, exception_debug=False):
   '''check ip function'''
   try:
      ipaddress.ip_address(pos_ip)
   exceptValueErroras e:
      if exception_debug: print(e)
      return(False)
   return(True)

check_ip(ip, exception_debug=True)

decarator

Элемент перед функцией, который реализует какую-то особенность обработки данной функции.

• • @staticmethod – по сути независимость метода от класса

@staticmethod — используется для создания метода, который ничего не знает о классе или экземпляре, через который он был вызван. Он просто получает переданные аргументы, без неявного первого аргумента, и его определение неизменяемо через наследование. Проще говоря, @staticmethod — это вроде обычной функции, определенной внутри класса, которая не имеет доступа к экземпляру, поэтому ее можно вызывать без создания экземпляра класса.

• • @timeout wrapt-timeout-decorator (ниже)

execution timeout

В python есть ряд разных реализаций execution timeout разной степени “поршивости” – что-то например не будет работать в Windows, что-то переодически подвисать из-за использования thread, вызова дочерних процессов и каких то еще проблем. Чаще всего работают через декоратор перед функцией. Наиболее стабильной функцией в этом контексте была timeout опция для subprocess и timeout утилита для linux – т.е. не питоновские “исходные” функции вовсе.

Способы, которые пробовал (первые более менее – последние совсем “зашквар”):

1. Способ wrapt-timeout-decorator работает в виде декоратора (перед функцией) и не имеет недостатков выше – возвращает как значения при успешном исполнении, не применяется на другие функции. Но есть недостаток в том, что процесс дочерний (напр. функция с timeout запускает subprocess) остается жив, что потенциально может приводить к формированию zombie.

pip3 install wrapt-timeout-decorator

import time
from wrapt_timeout_decorator import *
@timeout(5)

def mytest(message):
    print(message)
    for i in range(1,10):
        time.sleep(1)
        print('{} seconds have passed'.format(i))
    return "puk"

def mytest2(message):
    print(message)
    for i in range(1,10):
        time.sleep(1)
        print('{} seconds have passed'.format(i))
    return "puk"

res = mytest2('starting')
print(res)

2. timeout-decorator, не работает на Windows в режиме по умолчанию.

import time
import timeout_decorator

@timeout_decorator.timeout(5) # для Linux передаем timeout(5, use_signals=False)
def mytest():
  print("Start")
  for i in range(1,10):
    time.sleep(1) print("{} seconds have passed".format(i))
if __name__ == '__main__': mytest()

1. 3. Метод со stackoverflow работает и не требует установки модулей и зависимый библиотек (были с этим проблемы). Может зависать сам (ждет пользовательского ввода).

   #!/usr/bin/env python
   # -*- coding: utf-8 -*-
   from threading import Thread
   import functools
   
   def timeout(timeout):
   ...
   ...
   ..

import signal
import time

def handler(signum, frame):
   print("Forever is over!")
    raise Exception("end of time")

def loop_forever():
   while 1:
       print("sec")
        time.sleep(1)

def loop_forever2():
   while 1:
       print("sec2")
        time.sleep(1)

signal.signal(signal.SIGALRM, handler)
signal.alarm(10)
try:
  loop_forever2()
except:
    print("some shit")

5. Есть так же вариант с использованием thread, но он так же довольно кривой т.к. в таком случае нужно собирать output из процесса, например, используя shared variables, что усложняет код.

import multiprocessing
import time

# bar
def bar():
    for i in range(100):
        print("Tick")
        time.sleep(1)

# Start bar as a process
p = multiprocessing.Process(target=bar)
p.start()

# Wait for 10 seconds or until process finishes
p.join(10)

# If thred is still active
if p.is_alive():
    print("running... let's kill it...")
    # Terminate
    p.terminate()
    p.join()
p = multiprocessing.Process(target=bar)
p.start()

ООП

ПРОСТОЙ пример

Инициализируем SSH класс.

• • (# Assign params) Переменные назначаем из dictionary динамически по key:value, что позволяет не создавать одноименные параметры внутри класса – очень удобно.
  • (# Ssh login) Возвращаем объект SSH в котором уже пройдена аутентификация (результат внутри класса в self.ssh_conn, self.ssh_client доступен всем функциям класса). 

class SshControl:
  def __init__(self, kwargs):
    # Assign params
    for key, value in params.items():
      setattr(self, key, value)
    log.debug(f"{params}", self.log_fname)
    # Ssh login
    log.debug(f"[*] ssh_login to {self.ssh_ip} with {self.ssh_user}:{self.ssh_password}...", self.log_fname)
    self.ssh_client = paramiko.SSHClient()
    self.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    self.ssh_client.connect(self.ssh_ip, username=self.ssh_user, password=self.ssh_password)
    time.sleep(self.ssh_timeout_shell_waiting)
    self.ssh_conn = self.ssh_client.invoke_shell()

ssh_params = { "ip": "172.16.13.2", "ssh_user": "admin", "ssh_password": "admin", "port": 22, "log_fname": "test.log"}
ssh = SshControl(ssh_params)

Классы (CLasses) 

super – функция, которая может использоваться в базовом сценарии для переопределения функции, в наследовании.

Вызов метода внутри класса возможен с помощью self.

def json_gen(self, var1, var2, var3):
  ...

self.json_gen(var1, var2, var3)

Аналогично через self. можно передавать переменные между методами класса – не обязательно их передавать в сам метод в виде переменной.

self.somevar = "test"

OS.SYSTEM (EXTERNAL COMMAND)

С системным модулем subprocess запуск внешних команд похож на обычный shell или запуск в ruby/php через ‘command’. НО есть огромная разница – вывод не сохранить сразу в переменную в python, он только отображается в bash.

В целом общее мнение, что os.system сейчас лучше не использовать, а использовать subprocess.

import sys

os.system("ip netns exec VRF1 ip a | grep 'inet ' | awk '{print $NF}'")

Subprocess (external command)

• Бывают случаи, когда команды пересылаемые с рядом pipe linux должны запускаться не пользователем напрямую в терминале (напр. cron или запуск скрипта через WEB), в таких случаях чаще всего наиболее рабочим будет вариант запуска отдельного скрипта, включающего команды с pipe, а не попытка запуска всех pipe непосредственно в subprocess/os.system. Пример с настройкой прерываний через web скрипт.

subprocess.call( "sudo /var/www/cgi-bin/scr.sh %s" % iface, shell=True )

subprocess.check_output( "sudo /bin/grep -E %s /proc/interrupts | /usr/bin/awk '{print $1}' | /usr/bin/tr ':' ' ' | while read irq; do sudo echo 000002 > /proc/irq/$irq/smp_affinity; done" % iface, shell=True )

• С системным модулем subprocess запуск внешних команд похож gem Ruby Open3. Subprocess.run появился в новых версиях subprocess, раньше его не было. По практике – всегда, когда возможно, лучше использовать run вместо check_output.

import subprocess

subprocess.run(["ls", "-l"])
subprocess.check_output(["snmpget", "-v", "2c", "-c", "test", "-Ov", "-Oq", ip, oid])

• Для длинных команд удобно сначала использовать split и переменную.

cmd = "ip netns exec VRF1 ip a".split()
subprocess.run(cmd)

• Вывод STDOUT/STDERR скрыть можно через опции. Для check_output можно скрыть только stderr (stdout аргумент не разрешен для check_output).

import os
import subprocess

null = open(os.devnull, 'w')
subprocess.run(cmd, stderr=null, stdout=null)
subprocess.check_output(cmd, stderr=null)

• Так же может быть перенаправление stderr в stdout. Например, утилиты зачастую информацию о своей версии выводят в stderr.

tcpreplay_version_info = subprocess.check_output("tcpreplay -V".split(" "), stderr=subprocess.STDOUT)

write_to_file(tcpreplay_version_info, "file_with_tcpreplay_info.txt")

• Получение stdout/stderr в subprocess возможно и с использованием run и с использованием Popen при перенаправлении stdout в PIPE. При использовании check_output stdout изменить нельзя и он не отображается. По практике Popen лучше не использовать в пользу замены на run, несмотря на наличие в Popen функционала, которого нет в других режимах subpocess (напр. получение дочернего pid процесса при использовании preexec_fn=os.setsid). Проблема в том, что Popen может работать нестабильно (независимо от использования опций) и фризить периодически дочерний процесс.

test = subprocess.Popen(cmd, stdout=subprocess.PIPE)
data = test.communicate()
print(data)

test = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
print(data)
print(test.stdout) # вывод в виде byte cтроки
print(test.stdout.decode()) # преобразуем в string
print(test.returncode)

Обратите внимание на то, что когда вы запускаете этот код, цель которого – поддержка связи, он будет дожидаться финиша процесса, После чего выдаст кортеж, состоящий из двух элементов, которые являются содержимым stdout и stderr.

• Вместо преобразования byte строки через decode (как выше) можно указать кодирование вывода в unicode используя utf-8 сразу в subprocess.

test = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8')

Execution Timeout (subprocess)

В subprocess есть встроенная функция с execution timeout, которая работает с subprocess.run. 

res = subprocess.run("some_scripts", stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8', timeout=900)

Timeout + stdout/err in file (немного костылей)

Очень удобно использовать timeout, но есть недостаток – если мы присваиваем вывод в переменную, она не соберет данные при exception execution timeout. В итоге мы не будем знать что же сформировалось за время работы скрипта/не увидим ошибки. Решение – использовать запись в файл stdout/stderr через промежуточный скрипт bash скрипт. Обязательно /bin/bash, а не /bin/sh, иначе редирект не будет работать.

scripts.sh
#!/bin/bash
$1 $2 $3 &>$4

res = subprocess.run("some_scripts opt1 opt2 log_file.log", timeout=900)

Background

Запуск background subprocess.

test = subprocess.Popen(cmd)

Background + сбор вывода (еще немного костылей)

Для запуска в background со сбором вывода или при использовании в коде спец. символов (например, *|) может потребоваться использование shell.

test = subprocess.run("ls -ltr *py", shell=True)

Запуск в background + запись в файл ((есть так же вариант с промежуточный bash-скриптом, см выше)). Реализация через shell:

• tee для записи в файл,
• & для background,
• stdout=null для отброса вывода в консоль (запись только в файл)

import os
null = open(os.devnull, 'w')

subprocess.call('some_script.hz | tee log_file &', shell=True, stdout=null)

OUTLOOK (MAIL)

Отправка письма через outlook приложение на Windows, работает. Для нескольких адресатов указываем их через “;” в string.

import win32com.client as win32
outlook = win32.Dispatch('outlook.application')
mail = outlook.CreateItem(0)
mail.To = 'To address'
mail.Subject = 'Message subject'
mail.Body = 'Message body'
mail.HTMLBody = '<h2>HTML Message body</h2>' #this field is optional

# To attach a file to the email (optional):
attachment = "Path to the attachment"
mail.Attachments.Add(attachment)

mail.Send()

SYSTEM INFO

Библиотека sys.

sys.platform == "win32"

SOCKETS

• Расчет по памяти: примерно 64 GB для поднятия 10 МЛН сессий (CC). т.е. порядка 6.7 KB на сессию

Python как и Ruby имеет встроенную библиотеку socket, которая позволяет как принимать данные из сети, так и отправлять их. С ней можно поднять базовый TCP-сервер, отправить какой-то трафик (часто используется в exploit’ах) и проч.

Базовая работа с socket: import, create, connect, send, receive, close.

import socket

s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
connect=s.connect((HOST,PORT))
s.send(evil)
data = s.recv(4096)
s.close()

Пример включения и настройки отправки TCP keepalive для создаваемого socket. Так же timeout можно сделать безусловным на socket (без idle).

s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
s.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 60)
s.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 60)
s.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 30)

s.settimeout(30)

Пример системных (при установке socket из python не работают):
root@debian8:~# cat /proc/sys/net/ipv4/tcp_keepalive_intvl
75
root@debian8:~# cat /proc/sys/net/ipv4/tcp_keepalive_time
7200

Настройка MSS (уменьшаем с default 1460 до 96). Сделать еще меньше (48 байт) можно настроив mtu на интерфейсе. При попытке задать MSS через  TCP_MAXSEG может выдавать ошибку OSError: [Errno 22] Invalid argument.

s.setsockopt(socket.IPPROTO_TCP, socket.TCP_MAXSEG, 96)

# sysctl -a | grep -i tcp | grep mss 
net.ipv4.tcp_base_mss = 1024
net.ipv4.tcp_min_snd_mss = 48

Настройка размера TCP окна через задание размера буферов на прием и отправку. Про настройку Linux и особенности работы в TCP.

# python socket rewrite tcp_rmem/tcp_wmem (Linux) (не rmem_max/wmem_max)
s.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 2048) 
s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 2048)

# default from tcp_rmem/tcp_wmem (Linux) (не rmem_max/wmem_max)
>>> print(s.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF))
131072
>>> print(s.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF))
16384

Отключаем DF flag в IP-пакетах.

IP_MTU_DISCOVER = 10
IP_PMTUDISC_DONT = 0 # Never send DF frames.
s.setsockopt(socket.SOL_IP, IP_MTU_DISCOVER, IP_PMTUDISC_DONT)

Включаем возможность socket reuse, например полезно в сценарии когда демон постоянно по какой-то причине должен перезагружаться и “подниматься” максимально быстро (без ожидания полного высвобождения socket, аллокации памяти под сокет). Опцию можно использовать и для UDP сокета, в таком случае копию трафика будут получать все приложения, подключенные к socket, like a multicast/broadcast (справедливо и для multicast и для broadcast). Теория о socket reuse (в основном перевод статьи IBM) в статье TCP.

s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)


I'm aware that using SO_REUSEADDR with UDP in a *NIX environment, behaves like a multicast, where multiple clients bound to the same port can listen and receive broadcast datagrams simultaneously. 
Multiple UDP sockets on Windows bound to the same port will all receive broadcast packets together.

Отправка данных в TCP сокет.

s.send(data_new)

Прием данных в TCP. 4096 – количество байт, по факту столько одним пакетом редко когда будет получено из-за стандартного MTU 1500.

c.recv(4096)

https://www.ibm.com/docs/en/was-zos/8.5.5?topic=SS7K4U_8.5.5/com.ibm.websphere.nd.multiplatform.doc/ae/tprf_tunetcpip.html
https://stackoverflow.com/questions/36594400/what-is-backlog-in-tcp-connections

The default listen backlog is used to buffer spikes in new connections which come with a protocol like HTTP. The default listen backlog is 10 requests. You should use the TCP transport channel listenBacklog custom property to increase this value to something larger. For example:
listenBacklog=100

For channel types that are not HTTP, HTTP SSL, IIOP and IIOP SSL, the default for listenBacklog is 511.

https://stackoverflow.com/questions/2444459/python-sock-listen
You don't need to adjust the parameter to listen() to a larger number than 5. It specifies the number of unaccepted connections that the system will allow before refusing new connections. If not specified, a default reasonable value is chosen.

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.listen()

s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

socket.setblocking(flag)
Set blocking or non-blocking mode of the socket: if flag is false, the socket is set to non-blocking, else to blocking mode.
This method is a shorthand for certain settimeout() calls:
sock.setblocking(True) is equivalent to sock.settimeout(None)
sock.setblocking(False) is equivalent to sock.settimeout(0.0)

Примечания о тайм-аутах сокетов.
Объект сокета может находиться в одном из трех режимов: блокирующий, неблокирующий или тайм-аут. По умолчанию сокеты всегда создаются в режиме блокировки, но это можно изменить, вызвав функцию socket.setdefaulttimeout().
- В режиме блокировки (default), операции блокируются до тех пор, пока не будут завершены или система не возвратит ошибку (например, истекло время ожидания соединения). By default, TCP sockets are placed in a blocking mode. This means that the control is not returned to your program until some specific operation is complete. For example, if you call the connect() method, the connection blocks your program until the operation is complete.
- В неблокирующем режиме операции завершаются неудачно (с ошибкой, которая, к сожалению, зависит от системы), если они не могут быть выполнены немедленно: можно использовать функции из модуля select для определения того, когда и доступен ли сокет для чтения или записи.
- В режиме тайм-аута операции завершаются ошибкой, если они не могут быть завершены в течение тайм-аута, указанного для сокета (они вызывают исключение socket.timeout) или если система возвращает ошибку.

UDP socket (SOCK_DGRAM)

Особенности UDP socket. Несколько примеров.

TCP sockets should use socket.recv and UDP sockets should use socket.recvfrom. This is because TCP is a connection-oriented protocol. Once you create a connection, it does not change. UDP, on the other hand, is a connectionless ("send-and-forget") protocol. You use recvfrom so you know to whom you should send data back. Recvfrom does not work on TCP sockets the same way.

As for the 1024/2048, these represent the number of bytes you want to accept. Generally speaking, UDP has less overhead than TCP allowing you to receive more data, but this is not a strict rule and is almost negligible in this context. You can receive as much or as little as you would like. 4096 is common as well (for both).

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

if proto == "TCP":
  s.listen()
  c, addr = s.accept()
  print("TCP connection established with client ip/port: {}".format(addr))
  data = c.recv(4096)
else:
  data, addr = s.recvfrom(4096)
  print("UDP data received from client ip/port: {}".format(addr))
  data_new = "hello"
  s.sendto(data_new, addr)

Асинхронная обработка и сокеты

https://www.linux.org.ru/forum/development/14645644
Допустим у меня стоит задача опроса состояния GPIO-пина и выполнения некоторых действий в связи с этим.
Я могу поступить по-простому:
- завернуть пару-тройку for-while циклов, сделать в них poll состояния и логику того, что делать если состояние изменилось.
- А могу использовать asyncio. И я пока не понимаю, чем он отличается от такой же for-loops по большому счету, кроме того что представляет некоторую абстракцию над ними. Ну то есть читаемость - да, улучшается.

- asyncio в питоне это асинхронщина построенная на спрятанных тредах и мультиплексорах.  asyncio синтаксический сахар над ними. 
- asyncio просто аккуратно эти fd в epoll засунет. 
Идея в том, что мы делаем read/write над обычными fd, но с флагом NONBLOCK - который гарантирует, что процесс не войдёт в Sleep state, а вернёт errno = WOULDBLOCK. Дальше если всё что мы хотим читать/писать вернуло WOULDBLOCK - вручную вызываем sleep state на сразу множество fd. После просыпания ядро гарантирует (почти), что один из fd можно читать-писать без блокировки (есть данные в буфере). asyncio это всё автоматизирует синтаксическhttps://sysdev.me/python-asyncio/им сахаром до уровня как будто бы мы используем несколько тредов.

https://stackoverflow.com/questions/39145357/python-error-socket-error-errno-11-resource-temporarily-unavailable-when-s
In the server you are setting the remote socket (that returned by accept()) to non-blocking mode setblocking(0), which means that I/O on that socket will terminate immediately by an exception if there is no data to read.
There will usually be a period of time between establishing the connection with the server and the image data being sent by the client. The server attempts to immediately read data from the client once the connection is accepted, however, there might not be any data to read yet, so c.recv() raises a socket.error: [Errno 11] Resource temporarily unavailable exception. Errno 11 corresponds to EWOULDBLOCK, so recv() aborted because there was no data ready to read. Your code does not seem to require non-blocking sockets because there is an accept() at the top of the while loop, and so only one connection can be handled at a time. You can just remove the call to c.setblocking(0) and this problem should go away.

But what do I do, if the server is supposed to get multiple connection?
That's another problem. You can use select() to determine which of multiple sockets are ready to read, and then call recv() on those that are. You would also add the main server socket to the list of sockets passed to select() and call accept on that socket whenever select() indicates that it is readable. There are alternatives in the select module such as poll() et. al. You might also consider using the selectors module or asyncio if you are using Python 3.4 or later. asyncore is an option on Python 2.

NSLOOKUP

Простой nslooup на базе sockets.

simple nslookup in python

def simple_nslookup(fqdn):
  ip_list = []
  ais = socket.getaddrinfo(fqdn,0,0,0,0)
  for result in ais:
    if ":" not in result[-1][0]: # exclude IPv6
      ip_list.append(result[-1][0])
  return list(set(ip_list))

print(simple_nslookup("pool.ntp.org"))

IPADDRESS

Безусловно есть как поддержка IPv4Address/Network объектов, так и IPv6. Ниже описана работа только с IPv4. IPv6 по аналогии.

Address

С помощью функций модуля можно делать разные вещи, например с помощью функции ip_address легко проверять строки на соответствие формату IP-адреса.

import ipaddress

def check_ip(pos_ip):
   '''check ip function'''
   try:
      ipaddress.ip_address(pos_ip)
   except ValueError:
      return(False)
      return(True)

ip1 = "1.1.1.1"
ip2 = "2.2.2.2"
ip66 = "6.66.666.66"
print(ip1, check_ip(ip1))
print(ip2, check_ip(ip2))
print(ip66, check_ip(ip66))

1.1.1.1 True
2.2.2.2 True
6.66.666.66 False

С помощью функции is_private можем понять, относится ли IP к private диапазонам. Есть так же аналогичные функции is_multicast, is_loopback.

ip1 = "10.1.1.1"
ip1_validated = ipaddress.ip_address(ip1)
print(ip1_validated.is_private)
print(ip1_validated.is_multicast)
print(ip1_validated.is_loopback)

True
False
False

Можно так же сравнивать между собой адреса (ip1 > ip2), добавлять к адресам integer для получения нового адреса.

ip1 = "10.1.1.1"
ip1_validated = ipaddress.ip_address(ip1)
print(ip1_validated + 300)

10.1.2.45

Можно сделать проверку принадлежности Address объекта к сети Network.

ip = ipaddress.IPv4Address(u"1.1.1.1")
net = ipaddress.IPv4Network(u"1.1.1.1/255.255.255.0", strict=False)
print(ip in net)

True

Network

С помощью функций network_address и broadcast_address можно получить соответствующие адреса на основе адреса сети или даже IP адреса (strict false) для объекта Network. Так же можно получить общее количество адресов в сети, используя num_addresses. Так же Network является итерируемым объектом – из него можно извлечь объекты IPAddress для каждого адреса как по одному, так и по index.

net = ipaddress.IPv4Network(u"1.1.1.1/255.255.255.0", strict=False)
print('Network address:', net.network_address)
print('Broadcast:', net.broadcast_address)
print('Addr count:', net.num_addresses)

Network address: 1.1.1.0
Broadcast: 1.1.1.255
Addr count: 256


for ip in net:
  print(ip)

1.1.1.0
1.1.1.1
1.1.1.2
...
1.1.1.254
1.1.1.255


print(net[4])
1.1.1.4


# без strict будет ошибка, если указать не адрес сети
File "/usr/lib/python2.7/dist-packages/ipaddress.py", line 1662, in __init__
raise ValueError('%s has host bits set' % self)
ValueError: 1.1.1.1/24 has host bits set

Представлять маску в любом формате (with_netmask – mask, with_hostmask-wildcard, prefixlen – prefix) для объекта Network.

net = "10.1.1.0/24"
ipnet = ipaddress.ip_network(net)
print(ipnet.with_netmask)
print(ipnet.with_hostmask)
print(ipnet.prefixlen)

10.1.1.0/255.255.255.0
10.1.1.0/0.0.0.255
24

Можно так же для объекта Network использовать разделение сети на подсети – указав конкретную подсеть (new_prefix) или сделав вычитание из текущей (prefixlen_diff, по умолчанию 1, поэтому не обязательно указывать).

net = ipaddress.IPv4Network(u"1.1.1.1/255.255.255.0", strict=False)
print(list(net.subnets(new_prefix=27)))

[IPv4Network('1.1.1.0/27'), IPv4Network('1.1.1.32/27'), IPv4Network('1.1.1.64/27'), IPv4Network('1.1.1.96/27'), IPv4Network('1.1.1.128/27'), IPv4Network('1.1.1.160/27'), IPv4Network('1.1.1.192/27'), IPv4Network('1.1.1.224/27')]

print(list(net.subnets(prefixlen_diff=1)))
[IPv4Network('1.1.1.0/25'), IPv4Network('1.1.1.128/25')]

print(list(net.subnets()))
[IPv4Network('1.1.1.0/25'), IPv4Network('1.1.1.128/25')]

pyroute2 (netns)

Устанавливаем netns, возвращаемся в дефолтный. Работает с socket и psutil (напр. для просмотра сессий в VRF с использованием  psutil.net_connections) без каких-либо проблем.

pip3 install pyroute2
import socket
from pyroute2 import netns

netns.setns('VRF1')
# do some stuff in netns, than return back to default
netns.setns('/proc/1/ns/net')

SCAPy

• ScapyCheatSheet_v0.2 отсюда
• Official usage documentation

Самый удобный способ когда нужно сделать более чем одно действие по изменению сетевых дампов. Легко можно сделать даже непростые операции, например, заменить все сетевые (MAC, IP, PORT) адреса в дампе на другие, IPv6 header на IPv4. Замена IPv6 header на IPv4 подразумевает извлечение payload, создание IPv4 header, добавление в него payload, пересчет checksum и length (см. в HEX EDITORS).

Установка

sudo pip3 install scapy

Базовый запуск

from scapy.all import *

packets = rdpcap('test.pcap')
print(len(packets))
for packet in packets:
  print(packet) # print packet info from layer 2
  print(dir(packet))# print functions
  print(packet.payload) # print packet info from layer 3
  print(packet['TCP']) # print only TCP info
  print(packet['TCP'].flags) # print only TCP flag info

Scapy packet examples (часть из документации к TRex)

# PARSING
print(packet.summary())
print(packet.show())
packet['IP'].src
packet['IP'].src
packet['IP'].proto # определяем транспортный протокол 6 TCP, 17 UDP
packet['TCP'].sport
packet['TCP'].dport
packet['TCP'].flags == "S"

# Payload manipulation
payload = packet[TCP/UDP].payload # packet.payload is just a pointer to the next layer.
payload = raw(payload) # преобразуем payload в byte строку
if type(payload) == scapy.packet.Padding # проверяем не является ли payload == padding, иначе это пакет без фактического payload

# CREATION
# UDP header
Ether()/IP(src="16.0.0.1",dst="48.0.0.1")/UDP(dport=12,sport=1025)

# UDP over one vlan
Ether()/Dot1Q(vlan=12)/IP(src="16.0.0.1",dst="48.0.0.1")/UDP(dport=12,sport=1025)

# UDP QinQ
Ether()/Dot1Q(vlan=12)/Dot1Q(vlan=12)/IP(src="16.0.0.1",dst="48.0.0.1")/UDP(dport=12,sport=1025)

#TCP over IP over VLAN
Ether()/Dot1Q(vlan=12)/IP(src="16.0.0.1",dst="48.0.0.1")/TCP(dport=12,sport=1025)

# IPv6 over vlan
Ether()/Dot1Q(vlan=12)/IPv6(src="::5")/TCP(dport=12,sport=1025)

#Ipv6 over UDP over IP
Ether()/IP()/UDP()/IPv6(src="::5")/TCP(dport=12,sport=1025)

#DNS packet
Ether()/IP()/UDP()/DNS()

#HTTP packet
Ether()/IP()/TCP()/"GET / HTTP/1.1\r\nHost: www.google.com\r\n\r\n"

В целом scapy очень функциональный. Например, есть даже fuzzing (подробнее в документации по ссылке сверху).

>>> send(IP(dst="target")/fuzz(UDP()/NTP(version=4)),loop=1)

Поля можно посмотреть через ls. Очень удобно.

ls(Ether)
dst        : DestMACField                        = (None)
src        : SourceMACField                      = (None)
type       : XShortEnumField                     = (36864)

ls(IP)
version    : BitField (4 bits)                   = (4)
ihl        : BitField (4 bits)                   = (None)
tos        : XByteField                          = (0)
len        : ShortField                          = (None)
id         : ShortField                          = (1)
flags      : FlagsField (3 bits)                 = (<Flag 0 ()>)
frag       : BitField (13 bits)                  = (0)
ttl        : ByteField                           = (64)
proto      : ByteEnumField                       = (0)
chksum     : XShortField                         = (None)
src        : SourceIPField                       = (None)
dst        : DestIPField                         = (None)
options    : PacketListField                     = ([])

ls(TCP)
sport      : ShortEnumField                      = (20)
dport      : ShortEnumField                      = (80)
seq        : IntField                            = (0)
ack        : IntField                            = (0)
dataofs    : BitField (4 bits)                   = (None)
reserved   : BitField (3 bits)                   = (0)
flags      : FlagsField (9 bits)                 = (<Flag 2 (S)>)
window     : ShortField                          = (8192)
chksum     : XShortField                         = (None)
urgptr     : ShortField                          = (0)
options    : TCPOptionsField                     = (b'')

СКРИПТЫ

Скрипт по изменению MAC/IP адресов без создания нового пакета c пересчетом checksum IP/TCP.

from scapy.all import *

packets = rdpcap('/home/user/test.pcap')
new_packets = []
  for packet in packets:
 sport = packet['TCP'].sport
 payload = packet['TCP']
 if sport == 80 or sport == 8080:
  packet[Ether].dst = "00:10:11:a4:08:87"
  packet[Ether].src = "00:10:11:b0:8e:e7"
  packet[IP].dst = "172.0.0.2"
  packet[IP].src = "172.1.0.2"
 else:
  packet[Ether].dst = "00:10:11:a4:56:40"
  packet[Ether].src = "00:10:11:b0:8e:e5"
  packet[IP].dst = "172.1.0.2"
  packet[IP].src = "172.0.0.2"
del(packet[IP].chksum)
del(packet[TCP].chksum)
print(packet.show2()) # смотрим содержимое пакета
new_packets.append(packet)

wrpcap("test_new.pcap",new_packets)

Скрипт по извлечению TCP сегмента из существующего дампа, создание нового фрейма и IP пакета, добавление в него TCP payload. В итоге можно из исходного дампа удалить IPv6 header, изменить MAC/IP адреса и прочее, при этом сохранив полностью L4.

from scapy.all import *

packets = rdpcap('old.pcap')
new_packets = []
for packet in packets:
  sport = packet['TCP'].sport
  payload = packet['TCP']
  if sport == 49474:
    new_packet = Ether(src='00:11:22:33:44:55', dst='aa:bb:cc:dd:ee:ff')/IP(dst="1.1.1.1",src="192.168.0.1",proto=6)/Raw(load=payload)
  else:
    new_packet = Ether(src='00:11:22:33:44:55', dst='aa:bb:cc:dd:ee:ff')/IP(dst="192.168.0.1",src="1.1.1.1",proto=6)/Raw(load=payload)
  new_packets.append(new_packet)

wrpcap("new.pcap",new_packets)

Можно извлечь и данные других протоколов, например, ICMP. И добавить в новый IP пакет/фрейм.

payload = packet['ICMP']
new_packet = Ether(src='00:11:22:33:44:55', dst='aa:bb:cc:dd:ee:ff')/ IP(dst="1.1.1.1",src="172.17.0.2",proto=1)/ Raw(load=payload)

IP пакет с рандомным ID, установленным флагом DF, и TOS 16.

new_packet = Ether(src='00:11:22:33:44:55', dst='aa:bb:cc:dd:ee:ff')/ IP(dst="1.1.1.1",src="172.17.0.2",flags=2,id=random.randint(1,65534),tos=16,proto=6) / Raw(load=payload)

Модификация payload в дампе с пересчетом checksum.

TODO
48 54 54 50 2f 31 2e 31 20 32 30 30 20 4f 4b 0d HTTP/1.1 200 OK.
48 54 54 50 2f 31 2e 30 20 32 30 30 20 4f 4b 0d HTTP/1.0 200 OK.

from scapy.all import *

#!/usr/bin/env python3

# ver 1.0 by Petr V. Redkin

from scapy.all import *

def change_payload(packet, load):
  packet["Raw"].load = load
  del packet["IP"].len
  del packet["IP"].chksum
  del packet["TCP"].chksum
  return packet

packets = rdpcap('unmodified.pcap')
new_packets = []
n = 0
for packet in packets:
  payload = packet['TCP'].payload
  if "HTTP/1.1 200 OK" in str(payload):
    print("MATCH!")
    print("transaction changed!")
    new_payload = packet["Raw"].load.replace(b"HTTP/1.1 200 OK",b"HTTP/1.0 200 OK")
    packet = change_payload(packet,new_payload)
  new_packets.append(packet)

wrpcap("modified.pcap",new_packets)

Scapy отправка трафика

# Ethernet
new_packet = Ether(src='00:11:22:33:44:55', dst='aa:bb:cc:dd:ee:ff')/ IP(dst="172.17.0.2",src="1.2.3.4",proto=6) / Raw(load=payload)
sendp(new_packet, iface="enp5s0f0")

# IP
pi=IP(dst="10.13.37.218")/ICMP()
send(pi, loop=True)

Можно отправить и в vrf (так же можно напрямую в коде python используя pyroute2):

ip netns exec VRF1 python3 pcap_replay.py

Scapy может не распознавать дампы из-за некорректного magic в файле. Решается пересохранением дампа в Wireshark/tcpdump/tshark, которые этот дамп зачастую (рил кейс) распознают без проблем.

https://stackoverflow.com/questions/32465850/not-a-pcap-capture-file-bad-magic-scapy-python
"Not a pcap capture file (bad magic: %r)" % magic
scapy.error.Scapy_Exception: Not a pcap capture file (bad magic: b'4\xcd\xb2\xa1')

"Not a pcapng capture file (bad magic: %r)" % magic
scapy.error.Scapy_Exception: Not a pcapng capture file (bad magic: b'4\xcd\xb2\xa1')
scapy.error.Scapy_Exception: Not a supported capture file

Не надо называть скрипт, использующий scapy, по имени модуля (т.е. scapy.py).

https://stackoverflow.com/questions/59618380/modulenotfounderror-no-module-named-scapy

# python3 ./scapy.py
Traceback (most recent call last):
File "./scapy.py", line 1, in <module>
from scapy.all import * 
File "/opt/trex/trex-core/scripts/scapy.py", line 1, in <module> 
from scapy.all import * 
ModuleNotFoundError: No module named 'scapy.all'; 'scapy' is not a package 

pyshark

не умеет модифицировать пакеты. Scapy умеет.

sudo pip3 install pyshark

import pyshark
shark_cap = pyshark.FileCapture('test.pcap')
for packet in shark_cap:
  print(packet.tcp)
  print(dir(packet.tcp))

NTP client

import ntplib
from time import ctime

c = ntplib.NTPClient()
response = c.request('pool.ntp.org')
print(ctime(response.tx_time))

JSON

Python содержит встроенный модуль под названием json для кодирования и декодирования данных JSON.

import json

arr = { "test": 2, "test2": "xxx", "arr": [1,2,7] }
print(arr)
encoded = json.dumps(arr)
print(encoded)
decoded = json.loads(encoded)
print(decoded)

>
{'test': 2, 'test2': 'xxx', 'arr': [1, 2, 7]}
{"test": 2, "test2": "xxx", "arr": [1, 2, 7]}
{'test': 2, 'test2': 'xxx', 'arr': [1, 2, 7]}

Пример функции чтения json из файла (вместо loads используем load, который в качестве аргумента принимает файл).

d = get_json('strings.json')
print(d)

def get_json(j):
  with open(j) as f:
    j_parsed = json.load(f)
  return j_parsed

Пример выгрузки json с помощью urlib, парсинга его и сложения всех итемов “count”.

import json
from urllib.request import urlopen

url = "http://py4e-data.dr-chuck.net/comments_247944.json"
data = urlopen(url).read()
info = json.loads(data)

summ = 0
for item in info['comments']:
  print('Test', item['count'])
  summ = summ + int(item['count'])

print(summ)

Pretty print json

json_pattern = json.dumps(json_pattern, indent=4)

XML

You can use an XPath selector string to look through the entire tree of XML for any tag named 'count' with the following line of code:

counts = tree.findall('.//count')

from urllib.request import urlopen
import xml.etree.ElementTree as ET

url = input('Enter location: ')
data = urlopen(url).read()

print('Retrieved', len(data), 'characters')
print(data.decode())
tree = ET.fromstring(data)
results = tree.findall('.//count')

smm = 0
for item in results:
  print("Count: ", item.text)
  smm = smm + int(item.text)

print(smm)

JAML

Установка

sudo pip3 install PyYAML

import yaml

with open(some_file) as f:
  content = f.read()
test = yaml.safe_load(content)
print(test)

pPrint

Стандартный модуль. Позволяет читаемо отобразить list/dict и прочие данные. Единственное нужно помнить, что длинная строка (меньше width) может быть разделена на string literal – одна строка будет состоять из нескольких строк (каждая из которых укладывается в width), объединенных пробелом. При использовании на словари можно указывать уровень “погружения” depth. Подробнее тут.

from pprint import pprint

data = [('Interface', 'IP', 'Status', 'Protocol'),
 ('FastEthernet0/0', '15.0.15.1', 'up', 'up'),
 ('FastEthernet0/1', '10.0.12.1', 'up', 'up'),
 ('FastEthernet0/2', '10.0.13.1', 'up', 'up'),
 ('Loopback0', '10.1.1.1', 'up', 'up'),
 ('Loopback100', '100.0.0.1', 'up', 'up')]
print(data)
pprint(data)

[('Interface', 'IP', 'Status', 'Protocol'), ('FastEthernet0/0', '15.0.15.1', 'up', 'up'), ('FastEthernet0/1', '10.0.12.1', 'up', 'up'), ('FastEthernet0/2', '10.0.13.1', 'up', 'up'), ('Loopback0', '10.1.1.1', 'up', 'up'), ('Loopback100', '100.0.0.1', 'up', 'up')]

[('Interface', 'IP', 'Status', 'Protocol'),
('FastEthernet0/0', '15.0.15.1', 'up', 'up'),
('FastEthernet0/1', '10.0.12.1', 'up', 'up'),
('FastEthernet0/2', '10.0.13.1', 'up', 'up'),
('Loopback0', '10.1.1.1', 'up', 'up'),
('Loopback100', '100.0.0.1', 'up', 'up')]

Tabulate

Позволяет list/dict (в том числе многомерные) отобразить в красивом виде – с разделением по табам, html тегами, в виде таблиц и проч. Требует установки т.к. не стандартный модуль. Подробно тут.

sudo pip3 install tabulate

from tabulate import tabulate

data = [('Interface', 'IP', 'Status', 'Protocol'),
 ('FastEthernet0/0', '15.0.15.1', 'up', 'up'),
 ('FastEthernet0/1', '10.0.12.1', 'up', 'up'),
 ('FastEthernet0/2', '10.0.13.1', 'up', 'up'),
 ('Loopback0', '10.1.1.1', 'up', 'up'),
 ('Loopback100', '100.0.0.1', 'up', 'up')]
print(tabulate(data, headers='firstrow'))

Interface        IP         Status    Protocol
---------------  ---------  --------  ----------
FastEthernet0/0  15.0.15.1  up        up
FastEthernet0/1  10.0.12.1  up        up
FastEthernet0/2  10.0.13.1  up        up
Loopback0        10.1.1.1   up        up
Loopback100      100.0.0.1  up        up


print(tabulate(data, headers='firstrow', tablefmt='grid'))
+-----------------+-----------+----------+------------+
| Interface | IP | Status | Protocol |
+=================+===========+==========+============+
| FastEthernet0/0 | 15.0.15.1 | up | up |
+-----------------+-----------+----------+------------+
| FastEthernet0/1 | 10.0.12.1 | up | up |
+-----------------+-----------+----------+------------+
| FastEthernet0/2 | 10.0.13.1 | up | up |
+-----------------+-----------+----------+------------+
| Loopback0 | 10.1.1.1 | up | up |
+-----------------+-----------+----------+------------+
| Loopback100 | 100.0.0.1 | up | up |
+-----------------+-----------+----------+------------+

Parsing WEB/парсинг

Простейший get-аналог curl с входом из терминала в виде трех переменных и выводом в терминал.

import requests
import sys

one = sys.argv[1]
two = sys.argv[2]
three = sys.argv[3]

# Set the URL you want to webscrape from
url = 'http://127.0.0.1:2222/?text={}%{}%{}'.format(one, two, three)
# Connect to the URL
print(requests.get(url).text)

Можно оптимизировать используя join c разделителем % и отбросом через slice первого элемента, чтобы на входе были любое количество переменных.

delim = "%"
arg_str = delim.join(sys.argv[1:])

# Set the URL you want to webscrape from
url = 'http://127.0.0.1:2222/?text={}'.format(arg_str)

С utf-8 есть траблы. Вот как решаются:

# Узнать кодировку ответа
requests.get(url).encoding

# Использовать заданную кодировку на полученный ответ
requests.encoding = 'utf-8'

# Пример
res = requests.get(url)
print(res.encoding)
res.encoding = 'utf-8'
print(res.text)

BeautifulSoup with ssl disable

Примеры парсинга c bs4 HTTP

from urllib.request import urlopen
from bs4 import BeautifulSoup

url = input('Enter - ')
html = urlopen(url).read()
soup = BeautifulSoup(html, "html.parser")

# Retrieve all of the anchor tags
tags = soup('a')

2) Выгружаем все с тагом span, далее смотрим в содержимое этого тага и складываем содержимое во всех тагах

from urllib.request import urlopen
from bs4 import BeautifulSoup

url = input('Enter - ')
html = urlopen(url).read()
soup = BeautifulSoup(html, "html.parser")

# Retrieve all of the span tags
tags = soup('span')
summ = 0
for tag in tags:
  # Look at the parts of a tag
  #print('Contents:', tag.contents[0])
  summ = summ + int(tag.contents[0])

print(summ)

HTTPS требует добавления всего нескольких строк и изменения одной:

import ssl

# Ignore SSL certificate errors
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE

html = urlopen(url, context=ctx).read()

SSL

Standard library, under the hood использует OpenSSL.

Создание графиков (plotly)

• example
• example2

DEBUG/отладчики

Отладчики зачастую встроены в IDE (PyCharm).

Пример базового отладчика – pythontutor.com

Позволяет проследить ход исполнения программы, например, понять, как итерируются объекты.

Пример полноценного отладчика – pdb. Вызывать отладчик можно с помощью set-trace из кода самой программы.

import pdb; 

...

pdb.set-trace()

Ускорение работы

• Потоки python по умолчанию выполняются на одном ядре
• Для ускорения python можно почитать эту статью

GIL

GIL расшифровывается как Global Interpreter Lock (Глобальная блокировка интерпретатора), и его задача состоит в том, чтобы сделать интерпретатор CPython потокобезопасным.

Первый эффект GIL хорошо известен: несколько потоков Python не могут работать параллельно. Таким образом, многопоточная программа не ((ВСЕГДА)) будет быстрее, чем ее однопоточный эквивалент, даже на многоядерной машине.

• • Переход от одного потока к двум почти в 2 раза ускоряет выполнение, потому что потоки выполняются параллельно. Добавление дополнительных потоков не сильно помогает, потому что на моей машине всего два физических ядра. Вывод здесь заключается в том, что можно ускорить процессорно-интенсивный код Python с помощью многопоточности, если код вызывает функции C, которые освобождают GIL. Обратите внимание, что такие функции можно найти не только в стандартной библиотеке, но и в мощных сторонних модулях, таких как NumPy. Вы даже можете самостоятельно написать расширение C, которое выпустит GIL.
  • На самом деле многопоточные программы могут работать ((ДАЖЕ)) медленнее из-за накладных расходов, связанных с переключением контекста. Интервал переключения по умолчанию составляет 5 мс, поэтому переключение контекста происходит не так часто. Но если мы уменьшим интервал переключения, мы увидим замедление. 

GIL позволяет только одному потоку ОС выполнять байт-код Python в любой момент времени. Следствием этого является невозможность ускорить выполнение кода Python с интенсивным использованием процессора путем распределения работы между несколькими потоками.

Цикл выполнения байт-кода – это бесконечный цикл, который содержит гигантский switch всех возможных инструкций байт-кода. Чтобы запустить цикл, поток должен удерживать GIL. Время от времени поток должен приостанавливать исполнение байт-кода. Он проверяет, есть ли какие-либо причины для этого в начале каждой итерации цикла. Нас интересует одна из таких причин: другой поток запросил GIL.

• • В однопоточной программе на Python основной поток является единственным потоком, и он никогда не выпускает GIL.
  • Чтобы получить GIL, поток сначала проверяет, захвачен ли GIL каким-либо другим потоком. Если это не так, поток немедленно получает GIL. В противном случае он ждет, пока GIL не будет освобожден. Он ожидает фиксированного интервала времени, называемого интервалом переключения (по умолчанию 5 мс), и если GIL не будет выпущен в течение этого времени, он устанавливает флаги eval_breaker и gil_drop_request. Флаг eval_breaker указывает потоку, удерживающему GIL, приостановить выполнение байт-кода, а gil_drop_request объясняет, почему. Поток, удерживающий GIL, видит флаги, когда при запуске следующей итерации цикла, и освобождает GIL. Он уведомляет об этом потоки, ожидающие GIL, и один из них захватывает GIL. Какой именно поток получит GIL – зависит от операционной системы, так что это может быть поток, который установил флаги, а может быть и другой поток, также ожидающий GIL.

Кто-то активно пытается убрать GIL из Python

SQL

В Python в качестве стандартной библиотеки включен sqlite3 и его достаточно для основных/базовых задач. Базовый пример:

import sqlite3
con = sqlite3.connect('example.db')
cur = con.cursor()

# Create table
cur.execute('''CREATE TABLE stocks
(date text, trans text, symbol text, qty real, price real)''')

# Insert a row of data
cur.execute("INSERT INTO stocks VALUES ('2006-01-05','BUY','RHAT',100,35.14)")

# Save (commit) the changes
con.commit()

# Get data
for row in cur.execute('SELECT * FROM stocks ORDER BY price'):
  print(row)

# We can also close the connection if we are done with it.
# Just be sure any changes have been committed or they will be lost.
con.close()

What’s new in new python versions

some_funct(a,b,/):

раньше
  f'var={var}'
сейчас можно
  f'{var=}'
результат один
  'var=var_value'

ошибки

Ошибка

TabError: inconsistent use of tabs and spaces in indentation

Появляется если разные разделители (tab and spaces), нужно к одному привести (лучше в соответствии с pep8 4 spaces вместо каждого tab). Проще всего это сделать используя sublime text, у него есть полезная функция по конвертации в один формат – View -> Indentation -> Convert Indentation to tabs/spaces.

Меняем byte object на string

strng.decode()
#a bytes-like object is required, not 'str'

Меняем string на byte object

btobj.encode()
#cannot use a string pattern on a bytes-like object

Ошибка

UnboundLocalError: local variable 'exection_timeout' referenced before assignment

Появляется когда объявили глобальную переменную и захотели переназначить ее в методе. Можно исправить сделав ее глобальной, можно передать explicit в метод при его вызове и уже после этого переназначить внутри метода.

Не поддерживается except по ‘FileNotFoundError’, вместо него можно использовать IOError.

FileNotFoundError apparently doesn't exist in python 2.7 . a1ee437 should fix this.
except FileNotFoundError:
NameError: global name 'FileNotFoundError' is not defined

except IOError: