• Tweet

• Tweet

Coding

Basic template

import sys
#import traceback

def main():
    try:
        arguments = parse_arguments()
        process_data()
        print("Processed successfully:")
    except Exception as e:
        print(e)
        # traceback.print_exc(file=sys.stdout)
        sys.exit(1)
if __name__ == '__main__':
    main()

Parse arguments

import argparse
def parse_arguments():
    parser = argparse.ArgumentParser(description='Parsing arguments')
    parser.add_argument('--paramNAme', required=True, dest='fieldName')
    ...
    return parser.parse_args()

Private methods

Name to start by “__”

def __REPLACE_METHOD_NAME

Operators

• Negation: not
• Boolean comparison: is, is not, and, or
• String comparison: ==, !=, >, …
• String concatenation: +
• String contains: substring in container_string
• Membership: in, not in

Use functions from another file

import REPLACE_FILE_NAME
...
result = REPLACE_FILE_NAME.REPLACE_METHOD_NAME(xxx)

Print multiple values

print('REPLACE_TEXT %s xxx %s...' % (VALUE_1, VALUE_2...))

Clone an object

from copy import deepcopy
clone = deepcopy(ORIGINAL_OBJECT)

Sleep x seconds

import time
time.sleep(REPLACE_NUMBER_SECONDS)

Run shell commands

import subprocess
output = subprocess.check_output(command, shell=True).decode()

 Declare a list and add elements to it

my_list = []
my_list.append(REPLACE_OBJECT)

If-else-if

if xxx:
    ...
elif yyy:
    ...
else:
    ...

Exception management

try:
    REPLACE_CODE
except Exception as e:
    REPLACE_ACTION
...
if(REPLACE_CONDITION):
    REPLACE_SUCCESSFUL_ACTIONS
else:
    raise Exception("REPLACE_ERROR_MESSAGE")

Class declaration and object creation

# Class definition
class REPLACE_CLASS_NAME:
  def __init__(self, REPLACE_FIELD_ONE, REPLACE_FIELD_TWO, ...):
    self.REPLACE_FIELD_ONE = REPLACE_FIELD_ONE
    self.REPLACE_FIELD_TWO = REPLACE_FIELD_TWO
    ...
# Object creation
OBJECT_NAME = REPLACE_CLASS_NAME(REPLACE_FIELD_ONE, REPLACE_FIELD_TWO...)

Split a string and get the last element of the list

REPLACE_STRING_NAME.split("/")[-1]

File operations

Delete a file

import os
if os.path.exists(REPLACE_FILE_PATH):
  os.remove(REPLACE_FILE_PATH)
else:
  print("The file does not exist")

Copy a file

import shutil
shutil.copyfile(REPLACE_SOURCE, REPLACE_DESTINATION)

Write content into a file

Use “a” instead of “w” to append instead of replacing

f = open(REPLACE_FILE_PATH, "w")
f.write(REPLACE_CONTENT)
f.close()

Read a file’s content

file = open(REPLACE_FILE_PATH, "r")
return file.read()

Open a CSV file and parse its content

import csv
def parse_file(file):
    with open(file) as csv_file: 
        csv_reader = csv.reader(csv_file, delimiter=',')
        for row in csv_reader: 
            print("First column: " + row[0])

Read a properties file

import configparser
...
config = configparser.ConfigParser()
config.read(REPLACE_FILE_PATH)
value = config.get(REPLACE_SECTION_NAME, 'REPLACE_PROPERTY_NAME')

Sample properties file:

[DEFAULT]
PROPERTY_1.NAME=PROPERTY_VALUE_1
...
[REPLACE_SECTION_NAME]
PROPERTY_X.NAME=PROPERTY_VALUE_X
...

Security operations

Requests with OAuth2 authentication

from oauthlib.oauth2 import BackendApplicationClient 
from requests.auth import HTTPBasicAuth 
from requests_oauthlib import OAuth2Session
...
auth = HTTPBasicAuth(REPLACE_CLIENT_ID, REPLACE_CLIENT_SECRET)
client = BackendApplicationClient(client_id=REPLACE_CLIENT_ID, scopes=REPLACE_SCOPES)
auth_session = OAuth2Session(client=client)
auth_session.fetch_token(token_url=REPLACE_TOKEN_URL, auth=auth)
...
if request_type == 'get':
    result = auth_session.get(url=endpoint_url, headers=additional_headers)
elif request_type == 'delete':
    result = auth_session.delete(url=endpoint_url, data=request_body, headers=additional_headers)
elif request_type == 'post':
    result = auth_session.post(url=endpoint_url, data=request_body, headers=additional_headers)
elif request_type == 'put':
    result = auth_session.put(url=endpoint_url, headers=additional_headers)
else:
    raise Exception ('Unexpected request type: {}'.format(request_type))
# Check result.status_code is in valid range

Database operations

Query a database

import cx_Oracle as conn
connection = conn.connect(REPLACE_USER_ID, REPLACE_PASSWORD, REPLACE_DATABASE_NAME)
cursor = connection.cursor()
cursor.execute(REPLACE_QUERY)
connection.commit()
for result in cursor:
    print("First column: " + result[0])

Data processing with Pandas

import pandas as pd

Running scripts

Activate a venv

/usr/local/venv/REPLACE_VENV/bin/activate

Install packages via a proxy

sudo -H pip3 install --proxy 'http://REPLACE_USERNAME@REPLACE_PROXY_DNS_OR_IP:REPLACE_PROXY_PORT' REPLACE_PACKAGE_NAME