Python

"""Summary or Description of the Function

    Parameters:
    argument1 (int): Description of arg1

    Returns:
    int:Returning value
"""

__doc__

python -i file_name.py

python -m module_name

**kwargs

*args

variable_name: str

def function() -> str:

10_000

.5

input('Are you sure? (y/n): ').lower().strip()[:1]

def function_name(parameter_name,*, parameter_name)

def target_search(path, deep, *, depth = 0):
    target_dirs = list()

    if depth >= deep:
        return None

    depth += 1

    directories = os.listdir(path)

    if 'target_name' in directories:
        target_dirs.append(os.path.join(path, target_name))

    if not target_dirs:
        for directory in directories:
        next_path = os.path.join(path, directory)
    if os.path.isdir(next_path)
        result = target_search(next_path, deep, depth = depth)
    if result:
        target_dirs.extend(result)

def func():
    func.x = 'x'
    func.y = 'y'

print(func.x)

python -m pylint --generate-rcfile > .pylintrc

max(0, <vvariable>)

if x := True:
    print(x)

mode = dual
f"{mode=}"

def example():
    exammple.x = 1
    return x

example.x

fullness = 100
distance = 200
health = 70

conditions = [
    fullness == 100,
    distance == 200,
    health < 80,
]

if all(conditions):
    pass

if any(conditions):
    pass

dict1 = {"healt": 100, "distance": 50}
dict2 = {"health": 80, "enemys": 4}
dict3 = {**dict1, **dict2}

from itertools import chain

elements_list = [
    ['element1',],
    ['element1',],
    ['element1', 'element2',],
]

for element in chain.from_iterable(elements_list):
    print(element)

from typing import TypedDict

class SpaceShip(TypedDict, total=True):
    fullness: int
    health: int
    speed: int
    name: str

from ldap3 import Server, Connection, SUBTREE, MODIFY_ADD, MODIFY_REPLACE, MODIFY_ADD, MODIFY_DELETE

server = Server("ip_address")
connection = Connection(server, user="user_name", password="user_password")
connection.bind()
connection.unbind()

connection.search('search_base',
                    'search_filter',
                    SUBTREE,
                    attributes=['attributes_to_returned'])
conn.entries[0].returned_attributes

dn = json.loads(connection.entries[0].entry_to_json())['dn']
connection.modify(dn, {'user_attribute': [(MODIFY_REPLACE, [new_value])]})

dn = json.loads(connection.entries[0].entry_to_json())['dn']
connection.modify(dn, {'user_attribute': [(MODIFY_ADD, [new_value])]})

dn = json.loads(connection.entries[0].entry_to_json())['dn']
connection.modify(dn, {'user_attribute': [(MODIFY_DELETE, [delete_value])]})

import logging

logging.basicConfig(level=logging.logging_level, filename="path_to_log_file",
                      format='%(asctime)s %(process)d %(name)s %(levelname)s %(funcName)s %(message)s',
                      datefmt='%d-%b-%y %H:%M:%S')

logging.debug('')
logging.info('')
logging.critical('')
logging.error('')
logging.warning('')
logging.notset('')

stream=std.stdout

logging.disable(logging.ERROR)

def logging_configuration(logger):
    fh_formatter = logging.Formatter(fmt='%(asctime)s %(levelname)s: %(process)d %(name)s %(funcName)s %(message)s',
                                     datefmt='%m-%d-%Y %H:%M:%S')
    fh = logging.FileHandler(filename='<log_name>.log', delay=True)
    fh.setLevel(level=logging.INFO)
    fh.setFormatter(fh_formatter)

    sh_formatter = logging.Formatter(fmt='%(asctime)s %(process)d %(name)s %(levelname)s %(funcName)s %(message)s',
                                     datefmt='%d-%b-%y %H:%M:%S')
    sh = logging.StreamHandler()
    sh.setLevel(level=logging.INFO)
    sh.setFormatter(sh_formatter)

    logger.setLevel(logging.DEBUG)
    logger.addHandler(fh)
    logger.addHandler(sh)

loggger_interface = logging.getLogger('<logger_name>')
logging_configuration(loggger_interface)

import argparse

parser = argparse.ArgumentParser(description='', formatter_class=argparse.MetavarTypeHelpFormatter)
parser.add_argument('--example', dest='example', type=str, help='', default='')
parser.add_argument('--example2', dest='example2', type=bool, help='', action='store_true')
args = parser.parse_args()

args.example

import csv

with open('file_name', 'r') as csv_file:
csv_reader = csv.reader(csv_file)
    for row in csv_reader:
        row['column_name']

with open('file_name', 'w', newline='', encoding='windows-1251) as csv_file:
    fieldnames = ['column_name']
    csv_writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
    csv_writer.writeheader()
    csv_writer.writerow({'column_name': value})

import mysql.connector

mysql_connection = mariadb.connect(user='mysql_user', password='mysql_password', database='mysql_database', host='mysql_ip')
cursor = mysql_connection.cursor()
cursor.close()

cursor.execute("sql_query")
rows = cursor.fetchall()
rows[0].['column_name']

import configparser

config = configparser.ConfigParser()

config.read('configuration_file_name')
config.get('section_name', 'option_name', fallback=False)

[DEFAULT]
option_name = value

[section_name]
option_name = value

import xml.etree.ElementTree

root = ElementTree.Element('root_element')
sub_element = ElementTree.SubElement(root, 'sub_element')
sub_element.text = "sub_element_value"
ElementTree.dump(root)

tree = ElementTree.ElementTree(root)
tree.write('file_name', encoding="utf-8")

from queue import Queue

queue = Queue()

queue.put('element')

queue.get()

queue.task_done()

queue.join()

import threading

for i in range(5):
    name = f'Thread {i}'
    thread = CThread(name)
    thread.daemon = True
    thread.start()

Class CThread(threading.Thread):

    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        print(self.name)

class CThread(threading.Thread):

    def __init__(self, name, queue):
        threading.Thread.__init__(self)
        self.name = name
        self.queue = queue

    def run(self):
        self.queue.get()
        print(self.name)
        self.queue.task_done()

queue = Queue()

for i in range(5):
    name = f 'Thread {i}'
    thread = CThread(name, queue)
    thread.daemon = True
    thread.start()

queue.put('element')

import multiprocessing

Class CProcess(multiprocessing.Process):

    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        print(self.name)

import paramiko

ssh_key_object = io.StringIO(ssh_key)
ssh_rsa_key = paramiko.RSAKey.from_private_key(ssh_key_object, password_for_key)

ssh_rsa_key = paramiko.RSAKey.from_private_key_file(path_to_key, password_for_key)

client = paramiko.SSHClient()
client.load_system_host_keys()
client.connect(hostname='ip_address', port='port_number', username='user_name', password='user_password', pkey='key_file')
client.close()

sftp = client.open_sftp()
sftp.close()

stdin, stdout, stderr = client.exec_command("command")
stdout.read().decode()
stderr.read().decode()
stdout.chennel.recv_exit_status()

with open(key_file, encoding='utf-8-sig') as key_object:
    ssh_rsa_key = paramiko.RSAKey.from_private_key(key_object, password_for_key)

client.exec_command("command_line", shell=True)
client.exec_command(["command", 'args'], shell=False)

stdin, stdout, stderr = client..exec_command("sudo -u root -s /bin/bash")
stdin.write('ls\n')
stdin.flush()

import yaml

with open('file_name') as file:
    data = yaml.load(file, Loader=yaml.FullLoader)

yaml.full_load(file)

from jinja2 import Environment, FileSystemLoader

ENV = Environment(loader=FileSystemLoader('.'), trim_blocks=True, lstrip_blocks=True)
ENV = Environment(loader=FileSystemLoader('.'))

ENV.filters['function_name'] = function_name

template = ENV.get_template("template_name")

template.render(some_context=data)

import json
import requests

headers = {
        'Authorization': 'OAuth ',
        'Accept': 'application/json',
        'Content-Type': 'application/json'
    }

params = {
        'fields': '',
    }

data = {
        'fields': {'': ''}
    }

url = "https://site_name/"

response = requests.get(url, headers=headers, params=params)

http_auth = requests.auth.HTTPBasicAuth('login', 'password')
response = requests.get(url, auth=http_auth)

response = requests.post(url=url, headers=headers, params=params, data=json.dump(data))

response.json()
json.loads(response.text)
response.text
response.text.encode('utf8')
json.dumps(response.json(), indent=2)
json.loads(response.text)['antivirus_status']

response.status_code

response.headers

response.encoding

file = open(path_to_file, 'rb')

response = request.post(url=url, headers=headers, params=params, data=data, files=file)

response = request.post(url = url, stream = True)
with open(path_to_file, 'wb') as file:
    file.write(response.content)

with open(path_to_file, 'wb') as file:
    for chunk in file.iter_content(chunk_size=128):
        file.write(chunk)

from netmiko import ConnectHandler

device = ConnectHandler(host='ip_address', username='user_name', password='user_password', device_type='linux', secret='')

device.find_prompt()

device.enable(cmd='sudo -i', pattern='[sudo]')

device.send_command_expect()
device.send_command_timing()
device.send_command()

import os

import requests

url = 'http://nexus_address:8081/service/rest/v1/components'
http_auth = requests.auth.HTTPBasicAuth('login', 'password')

asset = open(path_to_file, 'rb')

params = {
    'repository': repository_name
}

payload = {
    'raw.asset1': asset
}

data = {
    'raw.directory': path_in_nexus,
    'raw.asset1.filename': file_name
}

response = request.post(url = url, auth = http_auth, params = params, data = data, files = payload)

asset.close()

url = 'http://nexus_address:8081/service/rest/v1/search/assets'
http_auth = requests.auth.HTTPBasicAuth('login', 'password')

params = {
    'repository': repository_name
}

response = request.get(url = url, auth = http_auth, params = params)
asset_info = response['items']
asset_url = asset_info['downloadUrl']

asset_object = request.get(url = asset_url, auth = http_auth)

with open(path_to_file, 'wb') as file:
    file.write(asset_object.content)

import tarfile

tar_file = tarfile.open(path_to_arhive. mode='w')
tar_file.add(path_to_file, arcname=path_to_file_in_arhive)
tar_file.close()

tar_file = tarfile.open(path_to_arhive, mode='r')
tar_file.extractall(path=path_to_directory)
tar_file.close()

import requests
import json

access_token = private_token

project_id = project id in GitLab

branch_name = branch name in project

variables = [{
    'key': variable_name,
    'value': variable_value,
}]

headers = {'PRIVATE-TOKEN': access_token, 'Content-Type': 'application/json'}

data = {'ref': branch_name, 'variables': variables}

url = f'https://gitlab_domain_name/api/v4/projects/{project_id}/pipeline'

response = requests.post(url=url, headers=headers, data=json.dumps(data), verify=False)

pipeline_info = response.json()

print(f'Pipeline running with parameters:\nURL: {pipeline_info["web_url"]}\nREF: {pipeline_info["ref"]}\nUSER: {pipeline_info["user"]["name"]}')

'|{}:<30|'.format(variable)

'|{}:>30|'.format(variable)

'|{}:^30|'.format(variable)

'|{}:*^30|'.format(variable)

file = io.BytesIO(<bytes>)

file = StringIO()
file.write(<data>)
file.getvalue().encode('utf-8')

try:
    result = 5 / 0
except ZeroDivisionError as message:
    print(message.args)
else:
    print()
finally:
    print()

class ExampleError(Exception):

    def __init__(self, message, input_data=None):
        self.message = message
        self.input_data = input_data

    def __str__(self):
        return self.message

raise ExampleError('example error')

def get_exception_example():
    return ValueError

exception_example = ValueError

try:
    pass
except exception_example:
    raise CustomException() from None

try:
    pass
except get_exception_example() as exc:
    raise CustomException() from exc

try:
    pass
except :
    raise

try:
    pass
except Exception:
    raise SyntaxError()

a = [i+10 for i in range(10)]

for index, value in range(10):
    a[index] = value + 10

a = [i:i+10 for i in range(10)]

a = (i+10 for i in range(10))

def func(number):
    for i in range(number):
        yield i + 10

        return / raise StopIteration()

a = func(10)
print(a.__next__())
print(next(a))

for i in a:
    print(i)

class Example:
    def __init__(self):
        self.i = 0

    def __iter__(self):
        self.i = 0
        return self

    def __next__(self):
        self.i += 1
        return self.i

        raise StopIteration()

lambda x: x + 1

class Multiplier:

    def __init__(self, n):
        self.n = n

    def __call__(self, x):
        return x * self.n

object = Multiplier(n=2)
object(x=2)

def multiplier(y):
    def multiply_column(x):
        return y * x

    return multiply_column

def time_track(func):
    @wraps(func)
    def wrapper(*args, **kwds)
      started_at = time.time()

      result = func(*args, **kwargs)

      ended_at = time.time()
      elapsed = round(ended_at - started_at, 4)
      print(elapsed)
      return result
    return wrapper

def some_function()
    pass

time_track(some_function)

@time_track
def some_function_v2()
    pass

from contextlib import contextmanager
import random


@contextmanager
def next_number(number):
    try:
        yield number + random.randint(0, 10)
    except Exception as exc:
        print(exc)


with next_number(1) as next_n:
    print(next_n)

current_date = datetime.datetime.now()
next_date = current_date + datetime.timedelta(days=1)
locale.setlocale(locale.LC_TIME, "ru_RU.utf8")
next_date.strftime('%a %d-%B')

import requests
import json

http_header = {
    'Accept': 'application/json',
    'User-Agent': '<agent_name>',
    'X-Requested-By': '<agent_name>'
}

auth = requests.ayth.HTTPBasicAuth(<user_name>, <user_password>)

url = http:<server_name>:8001/management/weblogic/latest/domainRuntime/serverLifeCycleRuntimes/<server_name>/forceShutdown
response = requests.post(url=url, headers=headers, auth=auth)

url = http:<server_name>:8001/management/weblogic/latest/domainRuntime/serverLifeCycleRuntimes/<server_name>/start
response = requests.post(url=url, headers=headers, auth=auth)

application_files = {
    'model': (None, json.dumps(dict())),
    'sourcePath': open(<path_to_application>, 'rb'),
    'planPath': open(<path_to_xml>, 'rb')
}

url = http:<server_name>:8001/management/weblogic/latest/edit/appDeployments/<application_name>/redeploy
response = requests.post(url=url, headers=headers, auth=auth, files=application_files)

from selenium import webdriver

profile = webdriver.FirefoxProfile()
profile.set_preference('intl.accept_languages', 'en-GB')
driver = webdriver.Firefox(firefox_profile=profile)

driver.get("http://www.google.com")
elem = driver.find_element_by_name("q")
elem.send_keys("Python")
elem.submit()
elem1 = driver.find_element_by_link_text('Sign in')
elem1.click()

def generate_wireguard_keys():
    """
    Generate a WireGuard private, public key and preshared key
    Requires that the 'wg' command is available on PATH
    Returns (private_key, public_key, psk), both strings
    https://techoverflow.net/2021/05/16/how-to-generate-wireguard-key-private-public-in-python-without-dependencies/
    https://github.com/pbengert/wireguard-config-generator/blob/main/wireguard-config-generator.py
    """
    privkey = subprocess.check_output(
        "wg genkey", shell=True).decode("utf-8").strip()
    pubkey = subprocess.check_output(
        f"echo '{privkey}' | wg pubkey", shell=True).decode("utf-8").strip()
    psk = subprocess.check_output(
        "wg genkey", shell=True).decode("utf-8").strip()
    return (privkey, pubkey, psk)

x_offset = 0
y_offset = 0

image1 = cv2.imread('image_1.jpg')
image2 = cv2.imread('image_2.jpg')

height, width, _ = image2.shape

roi = self.image1[y_offset:height+y_offset, x_offset:width+x_offset]
img2gray = cv2.cvtColor(image2, cv2.COLOR_BGR2GRAY)

ret, mask = cv2.threshold(img2gray, 10, 255, cv2.THRESH_BINARY)
mask_inv = cv2.bitwise_not(mask)

img1_bg = cv2.bitwise_and(roi, roi, mask=mask_inv)
img2_fg = cv2.bitwise_and(image2, image2, mask=mask)

dst = cv2.add(img1_bg, img2_fg)
image1[y_offset:height+y_offset, x_offset:width+x_offset] = dst

class ClassName:

    def __enter__(self):
        print()
        return self

    def __exit__(self):
        print()
        return True # disable crash when exception

with ClassName(self, exc_type, exc_val, exc_tb) as class_name:
    print()

class Person:
    def __init__(self, class_name, lvl):
        self._class_name = class_name
        self._lvl = lvl

    @classmethod
    def info(cls):
        print('info')

    @property
    def lvl(self):
        return self._lvl

    @property
    def class_name(self):
        return self._class_name

    @lvl.setter
    def lvl(self, lvl):
        self._lvl = lvl

    @class_name.setter
    def class_name(self, class_name):
        self._class_name = class_name


person = Person('druid', 80)
person.lvl = 70
print(person.lvl)

from abc import ABC, abstractmethod


class Person(ABC):

    @abstractmethod
    def attack(self):
        """"""

from dataclasses import dataclass

@dataclass(slots=False)
class Example:
    name: str
    email_id: str
    phone_number: str
    address: str

python -m cProfile <python file>

import pdb
pdb.set_trace()

pip install --upgrade pip

[build-system]
requires = ["setuptools>=60", "wheel"]
build-backend = "setuptools.build_meta"

import setuptools

with open("README.md", "r") as fh:
    long_description = fh.read()

setuptools.setup(
    name='<package name>',
    version='<package version>,
    description='',
    long_description='',
    long_description_content_type="text/markdown",
    author='<author name>',
    author_email='<author email>',
    maintainer='',
    maintainer_email='',
    url='<website url>',
    download_url='',
    packages=setuptools.find_packages(),
    py_modules=['<module name>',],
    classifiers= ['<describing>']
    license='Apache License 2.0',
    keywords=[],
    platforms='any',
    install_requires=[],
    python_requires='>=3.7',
    project_urls={},
)

pip install --upgrade build
pip install --upgrade wheel
pip install --upgrade setuptools

python -m build

pip install --upgrade twine

[distutils]
index-servers =
    gitlab
 
[gitlab]
repository = https://<gitlab domain>/api/v4/projects/<project id>/packages/pypi
username = <token name>
password = <token>

twine upload --repository gitlab dist/*

def example(x, y=None):
    if y is None:
        y = []
    x.append(x)
    return x

_variable_name = 'value'    # Protected
___variable_name = 'value'   # Private

from unicorn import *
from unicorn.x86_const import *
from unicorn.arm_const import *
from unicorn.mips_const import *

CODE = b"\x41\x4a\x66\x0f\xef\xc1"

mu = Uc(UC_ARCH_X86, UC_MODE_32)
mu.mem_map(0, <size of memmory>)
mu.mem_write(0, CODE)

mu.emu_start(0, len(CODE))

from capstone import *

CODE = b"\x41\x4a\x66\x0f\xef\xc1"

md = Cs(CS_ARCH_X86, CS_MODE_64)
for i in md.disasm(CODE, 0x1000):
    print("0x%x:\t%s\t%s" %(i.address, i.mnemonic, i.op_str))

import jwt
from pprint import pprint

token = input("Token: ")
jwt_structure = jwt.decode(token, algorithms=["RS256"], options={"verify_signature": False})
pprint(jwt_structure)

import os, sys, codecs
import time

R,W=os.pipe()
r,w=os.pipe()
print('R:', R, 'W:', W)
print('r', r, 'w:', w)
print('Parrent', os.getpid())
cpid = os.fork()
print('Child', cpid)
if cpid:
    os.dup2(0,100) # stdin > 100
    os.dup2(R,0) # R > stdin
    os.dup2(r,101) # r > 101
    os.close(R)
    os.close(r)
    os.close(W)
    os.close(w)