Python - usaging of "__init__.py"

create __init__.py file to get all files in the folder.

from os.path import dirname, basename, isfile, join
import glob

modules = glob.glob(join(dirname(__file__), "*.py"))
__all__ = [basename(f)[:-3] for f in modules if isfile(f) and not f.endswith('__init__.py')]

Python - scraping website with redirect links

This website (https://portalweb.cammesa.com/memnet1/Pages/descargas.aspx) has redirect link, and the post information is hidden in the response. you have to use the information to do the post, then you can get to the next step.

e.g.

from bs4 import BeautifulSoup
import requests
import zipfile


def find_between(s, first, last):
    try:
        start = s.index(first) + len(first)
        end = s.index(last, start)
        return s[start:end]
    except ValueError:
        return ""

headers = {
    'Content-Type': 'xxxxxxxxxxxx',    

    'User-Agent': 'xxxxxxxxxxx'}

session = requests.Session()

start_link = ''text = session.get(start_link, headers=headers).text
soup = BeautifulSoup(text, 'html.parser')

links = soup.find_all('a')

for link in links:
    if 'informe mensual' in str(link).lower():
        c_link = link.get('href')
        monthly_report_link_redirect = 'xxxxxxxx' + str(c_link)

        resp_informe_mensual = session.get(monthly_report_link_redirect, headers=headers)
        soup2_informe_mensual = BeautifulSoup(resp_informe_mensual.text, 'html.parser')

        # redirect to login page

        login_link = 'xxxxxxxxxxxxx'        

        data = {
            'Username': 'xxxxxx',            

            'Password': 'xxxxxx',            

            'RedirectTo': 'xxxxxxx',            

            'Remote_Addr': 'xxxxxxx'        

        }
        resp_open_frame_set = session.post(login_link, headers=headers, data=data)
        soup_open_frame_set = BeautifulSoup(resp_open_frame_set.text, 'html.parser')

        # final data page        data_link = 'xxxxxxxxxxxxx'

        resp_open_page = session.get(data_link, headers=headers)
        soup_open_page = BeautifulSoup(resp_open_page.text, 'html.parser')

        zip_link = soup_open_page.find_all('a')[2].get('href')
        full_link = 'xxxxxxx' + find_between(str(zip_link), '=/', 'zip') + 'zip'        

        file_name = find_between(str(zip_link), '$File/', 'zip') + 'zip'        

        print(full_link)
        print(file_name)

        # Download XXXX.zip file        

        r = requests.get(full_link,  auth=('xxxxxx', 'xxxxx'))
        with open(file_name, "wb") as code:
            code.write(r.content)

        # Unzip files        

        with zipfile.ZipFile(file_name, "r") as zip_ref:
            zip_ref.extractall()

Python - List of unique dictionaries

The usual way to find just the common elements in a set is to use Python's set class. Just add all the elements to the set, then convert the set to a list, and the duplicates are gone.

The problem, of course, is that a set() can only contain hashable entries, and a dict is not hashable.

If I had this problem, my solution would be to convert each dict into a string that represents the dict, then add all the strings to a set() then read out the string values as a list() and convert back to dict.

A good representation of a dict in string form is JSON format. And Python has a built-in module for JSON (called json of course).

Or, to make it work with Python 3.x (and recent versions of numpy), you need to convert array of dicts to numpy array of strings, e.g.

import numpy as np

list(np.unique(np.array(final_list).astype(str)))

SQL - Add constraint for an existing column

-----alter an existing column to have default value
  update  tbl_name
  set appended = getdate()

  alter table  tbl_name
  alter column appended DATETIME NOT NULL ;

  ALTER TABLE tbl_name
  ADD CONSTRAINT DF_Constraint DEFAULT GetDate() FOR appended;


-----get two digit month
 CONVERT(char(2), GetDate(), 101)

Python - get all the files in specified folders

This function below returns a list of file under current directory based on the file extension.

e.g. 

file_list = filebrowser(".zip")

def filebrowser(ext=""):
    "Returns files with an extension"    return [f for f in glob.glob(f"*{ext}")]

This function below return specified files in a list.

import os
from fnmatch import fnmatch

# 28 states

state_list = ['al', 'ar', 'ca', 'co', 'fl', 'id', 'ks', 'ky', 'la', 

'mi', 'mo', 'ms', 'mt', 'nd', 'ne', 'nm', 'nv', 'ny', 'oh', 'ok', 

'pa', 'sd', 'tn', 'tx', 'ut', 'va', 'wv', 'wy']

for state in state_list:
    root = os.path.join(os.path.dirname(os.path.abspath(__file__)), state)
    pattern = "*.py"    

    exclude_files = ['__init__', '_unit_test']
    for path, subdirs, files in os.walk(root):
        for name in files:
            if fnmatch(name, pattern) 

               and all(str(name).find(sub_name) == -1 for sub_name in exclude_files):

                print(os.path.join(path, name))

Python - Powerful eval() - concatenate string as function

eval(expression[, globals[, locals]])

The arguments are a string and optional globals and locals. If provided, globals must be a dictionary. If provided, locals can be any mapping object.

The expression argument is parsed and evaluated as a Python expression (technically speaking, a condition list) using the globals and locals dictionaries as global and local namespace. If the globals dictionary is present and does not contain a value for the key __builtins__, a reference to the dictionary of the built-in module builtins is inserted under that key before expression is parsed. This means that expression normally has full access to the standard builtins module and restricted environments are propagated. If the locals dictionary is omitted it defaults to the globals dictionary. If both dictionaries are omitted, the expression is executed with the globals and locals in the environment where eval() is called. Note, eval() does not have access to the nested scopes (non-locals) in the enclosing environment.

The return value is the result of the evaluated expression. Syntax errors are reported as exceptions. Example:

def schedule_job(file_name, scheduled_time):
    try:
        err_msg = pre_run_job(get_file(file_name))  

# Return error. if no error, then run the schedule        

        print(err_msg)

        logging.exception(err_msg)  

# Write error to log. If has error, write error to log without affecting other schedules        

        if err_msg == '':
            print('file is OK. Now start the schedule.')
            eval(scheduled_time)(call, ["python", get_file(file_name)])  

# schedule the job            

            run_schedule()
    except Exception as e:
        logging.exception(e)

Here, the argument 'scheduled_time' is actually a function like:

schedule.every(1).second.do()

But since I have lots of schedules to run, I'd rather like to write it into a function so that I don't need to write the whole thing lots of times.

By using eval(), it parses 'scheduled_time' as a function, then I can easily schedule jobs. e.g.

schedule_job('tx/permit/permit.py', 'schedule.every(1).seconds.do')

schedule_job('mt/permit/permit.py', 'schedule.every().monday.at("7:30").do')

Python - Capture error from one python script to another

When schedule a job,

•  you want to make sure the scheduled job runs successfully. 
• And you need to make sure that  schedules won't affect other schedules if all of them run in parallel. 
• You need some try catch statement to keep it running, in the mean while, keep track of the error. 

So you can use the following functions to do that. By doing so, the error of one job will be written in a log.txt file, in the same time, other jobs will keep running.

import functools
import os
import time
import logging
import schedule
from multiprocessing import *
from datetime import datetime
from subprocess import call, PIPE, Popen, TimeoutExpired
# initialize the log settings

logging.basicConfig(filename='schedule_permit.log', level=logging.INFO)

def pre_run_job(py_file):
    stderr = ''    cmd = ['python', py_file]
    p = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE)
    try:
        # let the job run 30 seconds to make sure it works fine
        stdout, stderr = p.communicate(timeout=30)  
    except TimeoutExpired:
        p.kill()  # stop the job after 30 seconds

    return stderr

def with_logging(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print('LOG: Running job "%s"' % func.__name__)
        print('Started at ' + str(datetime.now()))
        result = func(*args, **kwargs)
        print('LOG: Job "%s" Stopped' % func.__name__)
        print('Stopped at ' + str(datetime.now()))
        return result

    return wrapper

@with_logging
def run_test():
try:
    # Return error. if no error, then run the schedule
err_msg = pre_run_job(get_file('tx/permit/permit.py'))  
    print(err_msg) 
    logging.exception(err_msg)  
    if err_msg == '':
        print('file is OK. Now start the schedule.')
        schedule.every(1).seconds.do(call, ['python', get_file('tx/permit/permit.py')])
        run_schedule()
except Exception as e:
    logging.exception(e)
@with_loggingdef run_test2():
    try:
        schedule.every(10).seconds.do(print, 'hello')
        run_schedule()
    except Exception as e:
        logging.exception(e)


@with_loggingdef run_job():
    List_of_processes = [Process(target=run_test), Process(target=run_test2)]
    for p in List_of_processes:
        p.start()
    for p in List_of_processes:
        p.join()


if __name__ == "__main__":
    run_job()