<commit_before>#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# FOGLAMP_BEGIN
# See: http://foglamp.readthedocs.io/
# FOGLAMP_END
"""
Description: Based on FOGL-200 (https://docs.google.com/document/d/1GdMTerNq_-XQuAY0FJNQm09nbYQSjZudKJhY8x5Dq8c/edit) 
    the purge process is suppose to remove data based on either a user_id, or an X amount of time back depending on
    whether or not the configuration (config.json) requires to retain data that has not been sent to a  historian.
    
    As of now, all dependencies, with the exception of the database layer have been settled, and being used. These are
    -> configurations 
    -> retrieval of last ID sent to the Historian
     
     As for the database layer, the code currently uses SQLAlchemy, but can easily be switched out to use some other 
    tool to communicate with the database. 
    
     Once the purge process is called by the scheduler, it does the following: 
     1. Connects to the database 
     2. Retrieve information from the Configuration, and last ID sent to the historian
     3. Calculates how many rows are in the database to this point (using "NOW()") 
     4. Either DELETE rows based off age, or lastID (as well as "NOW()")
     5. Calculate necessary information regarding the purge process 
        -> total_rows_removed
        -> total_unsent_rows
        -> total_rows_remaining
        -> total_failed_to_remove
     6. INSERT information into log table 
     
     There currently isn't a formal confirmation that the purge process has succeeded, HOWEVER if 
     total_failed_to_remove > 0 then it is safe to assume that that there was an error with INSERTS, and if 
     total_failed_to_remove > total_rows_removed then PURGE completely failed. 
"""

import asyncio
import datetime
import random
import sqlalchemy
import sqlalchemy.dialects.postgresql
import time
from foglamp import configuration_manager
from foglamp import statistics

"""Script information and connection to the Database
"""
__author__ = "<NAME>"
__copyright__ = "Copyright (c) 2017 OSI Soft, LLC"
__license__ = "Apache 2.0"
__version__ = "${VERSION}"

# Create Connection
__CONNECTION_STRING = "postgres:///foglamp"

_DEFAULT_PURGE_CONFIG = {
    "age": {
        "description": "Age of data to be retained, all data that is older than this value will be removed," +
                       "unless retained. (in Hours)",
        "type": "integer",
        "default": "72"
    },
    "retainUnsent": {
        "description": "Retain data that has not been sent to any historian yet.",
        "type": "boolean",
        "default": "False"
    }
}
_CONFIG_CATEGORY_NAME = 'PURGE_READ'
_CONFIG_CATEGORY_DESCRIPTION = 'Purge the readings table'


"""Utilized tables 
"""
# Table purge against
# The pruge process utilizes only either id or ts respectively
_READING_TABLE = sqlalchemy.Table('readings', sqlalchemy.MetaData(),
                                  sqlalchemy.Column('id', sqlalchemy.BIGINT, primary_key=True),
                                  sqlalchemy.Column('asset_code', sqlalchemy.VARCHAR(50)),
                                  sqlalchemy.Column('read_key', sqlalchemy.dialects.postgresql.UUID,
                                                    default='00000000-0000-0000-0000-000000000000'),
                                  sqlalchemy.Column('reading', sqlalchemy.dialects.postgresql.JSON, default='{}'),
                                  sqlalchemy.Column('user_ts', sqlalchemy.TIMESTAMP(6),
                                                    default=sqlalchemy.func.current_timestamp()),
                                  sqlalchemy.Column('ts', sqlalchemy.TIMESTAMP(6),
                                                    default=sqlalchemy.func.current_timestamp()))


"""log table column information
id - the row id
code - process being logged 
level - Whether or not process succeeded (0 Success | 1 Failure | 2 Warning | 4 Info)
log - values being logged 
ts - current timestamp 
"""
# For 'ts' need need to have "onupdate=sqlalchemy.func.current_timestamp()" otherwise code returns sqlalchemy.exc.IntegrityError 
_LOG_TABLE = sqlalchemy.Table('log', sqlalchemy.MetaData(),
                                        sqlalchemy.Column('id', sqlalchemy.BIGINT, primary_key=True, autoincrement=True),
                                        sqlalchemy.Column('code', sqlalchemy.CHAR(5), default='PURGE'),
                                        sqlalchemy.Column('level', sqlalchemy.SMALLINT, default=0),
                                        sqlalchemy.Column('log', sqlalchemy.dialects.postgresql.JSONB, default={}),
                                        sqlalchemy.Column('ts', sqlalchemy.TIMESTAMP(6),
                                                          default=sqlalchemy.func.current_timestamp(),
                                                          onupdate=sqlalchemy.func.current_timestamp()))


"""Methods that support the purge process.  
"""


def get_nth_id() -> int:
    """Update the config file to have row ID somewhere within the oldest 100 rows.
    This method would potentially be replaced by a mechanism that is aware what was the last value sent to the
     historian.
    Returns:
        Method doesn't return anything
    """
    rand = random.randint(1, 100)
    stmt = "SELECT id FROM (SELECT id FROM readings ORDER BY id ASC LIMIT %s)t ORDER BY id DESC LIMIT 1"
    row_id = execute_command(stmt % rand).fetchall()
    try:
        return int(row_id[0][0])
    except IndexError:
        return 1


def convert_timestamp(set_time: str) -> datetime.timedelta:
    """Convert "age" in config file to timedelta. If only an integer is specified,  then 
        the code assumes that it is already in minutes (ie age:1 means wait 1 minute) 
    Args:
        set_time (str): Newest amount of  time back to delete
    Returns:
        converted set_time to datetime.timedelta value
    """
    if type(set_time) is int or set_time.isdigit():
        return datetime.timedelta(minutes=int(set_time))
    time_dict = {}
    tmp = 0

    for value in set_time.split(" "):
        if value.isdigit() is True:
            tmp = int(value)
        else:
            time_dict[value] = tmp

    time_in_sec = datetime.timedelta(seconds=0)
    time_in_min = datetime.timedelta(minutes=0)
    time_in_hr = datetime.timedelta(hours=0)
    time_in_day = datetime.timedelta(days=0)

    for key in time_dict.keys():
        if 'sec' in key:
            time_in_sec = datetime.timedelta(seconds=time_dict[key])
        elif 'min' in key:
            time_in_min = datetime.timedelta(minutes=time_dict[key])
        elif ('hr' in key) or ('hour' in key):
            time_in_hr = datetime.timedelta(hours=time_dict[key])
        elif ('day' in key) or ('dy' in key):
            time_in_day = datetime.timedelta(days=time_dict[key])
    return time_in_sec+time_in_min+time_in_hr+time_in_day


def execute_command(stmt: str):
    """Imitate connection to postgres that returns result.    
    Args:
        stmt (str): generated SQL query   
    Returns:
        Returns result set 
    """
    engine = sqlalchemy.create_engine(__CONNECTION_STRING, pool_size=20, max_overflow=0)
    conn = engine.connect()
    query_result = conn.execute(stmt)
    return query_result


"""The actual purge process 
"""


def purge(config, table_name) -> (int, int):
    """Column information
    start_time - time that purge process began
    end_time - time that purge process ended
    -------------------------------------------
    total_rows_removed - number of rows removed 
        - DELETE stmt returns value
    ------------------------------------------
    unsent_rows_removed - number of rows that weren't sent to historian, but were removed
    failed_removal - number of rows that failed to remove
        - SELECT stmt of DELETE
    rows_remaining - total number of rows remain at screenshot 
        - total_row - total_rows_removed
    """
    start_time = time.strftime('%Y-%m-%d %H:%M:%S.%s', time.localtime(time.time()))

    unsent_rows_removed = 0 
    total_rows_removed = 0
    last_id = get_nth_id()
    error_level = 0
    
    # Calculate current count and age_timestamp
    age_and_count_query = sqlalchemy.select([sqlalchemy.func.current_timestamp() - datetime.timedelta(hours=int(config['age']['value'])), sqlalchemy.func.count()]).select_from(table_name)
    result = execute_command(age_and_count_query).fetchall()
    age_timestamp = result[0][0]
    total_count = result[0][1]

    # MAX possible ID to delete
    max_id_query = sqlalchemy.select([sqlalchemy.func.max(table_name.c.id)]).select_from(table_name).where(table_name.c.user_ts <= age_timestamp)
    max_id = execute_command(max_id_query).fetchall()[0][0]

    # if retainUnsent is True then delete by last_id,  else delete by age
    if config['retainUnsent']['value'] == "True":
        delete_query = sqlalchemy.delete(table_name).where(table_name.c.id <= last_id)
        total_rows_removed = execute_command(delete_query).rowcount
        failed_removal_query = sqlalchemy.select([sqlalchemy.func.count()]).select_from(table_name).where(table_name.c.id <= last_id) 
        failed_removal = execute_command(failed_removal_query).fetchall()[0][0]
    
    else: 
        delete_query = sqlalchemy.delete(table_name).where(table_name.c.user_ts <= age_timestamp)
        total_rows_removed = execute_command(delete_query).rowcount
        failed_removal_query = sqlalchemy.select([sqlalchemy.func.count()]).select_from(table_name).where(table_name.c.user_ts <= age_timestamp)
        failed_removal = execute_command(failed_removal_query).fetchall()[0][0]
        unsent_rows_removed = int(max_id) - last_id
        if unsent_rows_removed < 0: 
            unsent_rows_removed = 0
    
    # Rows remaining is based on the snapshot taking at the start of the process 
    rows_remaining = int(total_count) - int(total_rows_removed)

    """Error Levels: 
    - 0: No errors
    - 1: Rows failed to remove 
    - 2: Unsent rows were removed
    """ 
    if  failed_removal > 0: 
        error_level = 1
    elif unsent_rows_removed > 0: 
        error_level = 2

    end_time = time.strftime('%Y-%m-%d %H:%M:%S.%s', time.localtime(time.time()))

    insert_into_log(level=error_level, log={"start_time":start_time, "end_time": end_time, "rowsRemoved": total_rows_removed, "unsentRowsRemoved": unsent_rows_removed,
        "failedRemovals": failed_removal, "rowsRemaining": rows_remaining})  

    return total_rows_removed, unsent_rows_removed 

def insert_into_log(level=0, log=None):
    """INSERT into log table values"""
    stmt = _LOG_TABLE.insert().values(code='PURGE', level=level, log=log)
    execute_command(stmt)


def purge_main():
    event_loop = asyncio.get_event_loop()
    event_loop.run_until_complete(configuration_manager.create_category(_CONFIG_CATEGORY_NAME, _DEFAULT_PURGE_CONFIG,
                                                                        _CONFIG_CATEGORY_DESCRIPTION))
<commit_msg>updated code to use MAX(foglamp.streams.last_object) in order to retrive neweset id sent to OMF<commit_after>#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# FOGLAMP_BEGIN
# See: http://foglamp.readthedocs.io/
# FOGLAMP_END
"""
Description: Based on FOGL-200 (https://docs.google.com/document/d/1GdMTerNq_-XQuAY0FJNQm09nbYQSjZudKJhY8x5Dq8c/edit) 
    the purge process is suppose to remove data based on either a user_id, or an X amount of time back depending on
    whether or not the configuration (config.json) requires to retain data that has not been sent to a  historian.
    
    As of now, all dependencies, with the exception of the database layer have been settled, and being used. These are
    -> configurations 
    -> retrieval of last ID sent to the Historian
     
     As for the database layer, the code currently uses SQLAlchemy, but can easily be switched out to use some other 
    tool to communicate with the database. 
    
     Once the purge process is called by the scheduler, it does the following: 
     1. Connects to the database 
     2. Retrieve information from the Configuration, and last ID sent to the historian
     3. Calculates how many rows are in the database to this point (using "NOW()") 
     4. Either DELETE rows based off age, or lastID (as well as "NOW()")
     5. Calculate necessary information regarding the purge process 
        -> total_rows_removed
        -> total_unsent_rows
        -> total_rows_remaining
        -> total_failed_to_remove
     6. INSERT information into log table 
     
     There currently isn't a formal confirmation that the purge process has succeeded, HOWEVER if 
     total_failed_to_remove > 0 then it is safe to assume that that there was an error with INSERTS, and if 
     total_failed_to_remove > total_rows_removed then PURGE completely failed. 
"""

import asyncio
import datetime
import random
import sqlalchemy
import sqlalchemy.dialects.postgresql
import time
from foglamp import configuration_manager
from foglamp import statistics

"""Script information and connection to the Database
"""
__author__ = "<NAME>"
__copyright__ = "Copyright (c) 2017 OSI Soft, LLC"
__license__ = "Apache 2.0"
__version__ = "${VERSION}"

# Create Connection
__CONNECTION_STRING = "postgres:///foglamp"

_DEFAULT_PURGE_CONFIG = {
    "age": {
        "description": "Age of data to be retained, all data that is older than this value will be removed," +
                       "unless retained. (in Hours)",
        "type": "integer",
        "default": "72"
    },
    "retainUnsent": {
        "description": "Retain data that has not been sent to any historian yet.",
        "type": "boolean",
        "default": "False"
    }
}
_CONFIG_CATEGORY_NAME = 'PURGE_READ'
_CONFIG_CATEGORY_DESCRIPTION = 'Purge the readings table'


"""Utilized tables 
"""
# Table purge against
# The pruge process utilizes only either id or ts respectively
_READING_TABLE = sqlalchemy.Table('readings', sqlalchemy.MetaData(),
                                  sqlalchemy.Column('id', sqlalchemy.BIGINT, primary_key=True),
                                  sqlalchemy.Column('asset_code', sqlalchemy.VARCHAR(50)),
                                  sqlalchemy.Column('read_key', sqlalchemy.dialects.postgresql.UUID,
                                                    default='00000000-0000-0000-0000-000000000000'),
                                  sqlalchemy.Column('reading', sqlalchemy.dialects.postgresql.JSON, default='{}'),
                                  sqlalchemy.Column('user_ts', sqlalchemy.TIMESTAMP(6),
                                                    default=sqlalchemy.func.current_timestamp()),
                                  sqlalchemy.Column('ts', sqlalchemy.TIMESTAMP(6),
                                                    default=sqlalchemy.func.current_timestamp()))


"""log table column information
id - the row id
code - process being logged 
level - Whether or not process succeeded (0 Success | 1 Failure | 2 Warning | 4 Info)
log - values being logged 
ts - current timestamp 
"""
# For 'ts' need need to have "onupdate=sqlalchemy.func.current_timestamp()" otherwise code returns sqlalchemy.exc.IntegrityError 
_LOG_TABLE = sqlalchemy.Table('log', sqlalchemy.MetaData(),
                                        sqlalchemy.Column('id', sqlalchemy.BIGINT, primary_key=True, autoincrement=True),
                                        sqlalchemy.Column('code', sqlalchemy.CHAR(5), default='PURGE'),
                                        sqlalchemy.Column('level', sqlalchemy.SMALLINT, default=0),
                                        sqlalchemy.Column('log', sqlalchemy.dialects.postgresql.JSONB, default={}),
                                        sqlalchemy.Column('ts', sqlalchemy.TIMESTAMP(6),
                                                          default=sqlalchemy.func.current_timestamp(),
                                                          onupdate=sqlalchemy.func.current_timestamp()))


"""Methods that support the purge process.  
"""


def get_last_id() -> int:
    """
    From foglamp.streams retrive the heighest last_object that corresponds to the newest readings table row id sent to Pi/OMF 
    Returns: 
        The newest `readings` table row id that was sent to Pi/OMF
    """
   stmt = "SELECT MAX(last_object) FROM foglamp.streams;" 
   return int(conn.execute(stmt).fetchall()[0][0]) 

def execute_command(stmt: str):
    """Imitate connection to postgres that returns result.    
    Args:
        stmt (str): generated SQL query   
    Returns:
        Returns result set 
    """
    engine = sqlalchemy.create_engine(__CONNECTION_STRING, pool_size=20, max_overflow=0)
    conn = engine.connect()
    query_result = conn.execute(stmt)
    return query_result


"""The actual purge process 
"""


def purge(config, table_name) -> (int, int):
    """Column information
    start_time - time that purge process began
    end_time - time that purge process ended
    -------------------------------------------
    total_rows_removed - number of rows removed 
        - DELETE stmt returns value
    ------------------------------------------
    unsent_rows_removed - number of rows that weren't sent to historian, but were removed
    failed_removal - number of rows that failed to remove
        - SELECT stmt of DELETE
    rows_remaining - total number of rows remain at screenshot 
        - total_row - total_rows_removed
    """
    start_time = time.strftime('%Y-%m-%d %H:%M:%S.%s', time.localtime(time.time()))

    unsent_rows_removed = 0 
    total_rows_removed = 0
    last_id = get_last_id()
    error_level = 0
    
    # Calculate current count and age_timestamp
    age_and_count_query = sqlalchemy.select([sqlalchemy.func.current_timestamp() - datetime.timedelta(hours=int(config['age']['value'])), 
        sqlalchemy.func.count()]).select_from(table_name)
    result = execute_command(age_and_count_query).fetchall()
    age_timestamp = result[0][0]
    total_count = result[0][1]

    # MAX possible ID to delete
    max_id_query = sqlalchemy.select([sqlalchemy.func.max(table_name.c.id)]).select_from(table_name).where(table_name.c.user_ts <= age_timestamp)
    max_id = execute_command(max_id_query).fetchall()[0][0]

    # if retainUnsent is True then delete by last_id,  else delete by age
    if config['retainUnsent']['value'] == "True":
        delete_query = sqlalchemy.delete(table_name).where(table_name.c.id <= last_id)
        total_rows_removed = execute_command(delete_query).rowcount
        failed_removal_query = sqlalchemy.select([sqlalchemy.func.count()]).select_from(table_name).where(table_name.c.id <= last_id) 
        failed_removal = execute_command(failed_removal_query).fetchall()[0][0]
    
    else: 
        delete_query = sqlalchemy.delete(table_name).where(table_name.c.user_ts <= age_timestamp)
        total_rows_removed = execute_command(delete_query).rowcount
        failed_removal_query = sqlalchemy.select([sqlalchemy.func.count()]).select_from(table_name).where(table_name.c.user_ts <= age_timestamp)
        failed_removal = execute_command(failed_removal_query).fetchall()[0][0]
        unsent_rows_removed = int(max_id) - last_id
        if unsent_rows_removed < 0: 
            unsent_rows_removed = 0
    
    # Rows remaining is based on the snapshot taking at the start of the process 
    rows_remaining = int(total_count) - int(total_rows_removed)

    """Error Levels: 
    - 0: No errors
    - 1: Rows failed to remove 
    - 2: Unsent rows were removed
    """ 
    if  failed_removal > 0: 
        error_level = 1
    elif unsent_rows_removed > 0: 
        error_level = 2

    end_time = time.strftime('%Y-%m-%d %H:%M:%S.%s', time.localtime(time.time()))

    insert_into_log(level=error_level, log={"start_time":start_time, "end_time": end_time, "rowsRemoved": total_rows_removed, "unsentRowsRemoved": 
        unsent_rows_removed, "failedRemovals": failed_removal, "rowsRemaining": rows_remaining})  

    return total_rows_removed, unsent_rows_removed 

def insert_into_log(level=0, log=None):
    """INSERT into log table values"""
    stmt = _LOG_TABLE.insert().values(code='PURGE', level=level, log=log)
    execute_command(stmt)


def purge_main():
    event_loop = asyncio.get_event_loop()
    event_loop.run_until_complete(configuration_manager.create_category(_CONFIG_CATEGORY_NAME, _DEFAULT_PURGE_CONFIG,
                                                                        _CONFIG_CATEGORY_DESCRIPTION))

{'fraction_non_alphanumeric': 0.1004381694255112, 'fraction_numerical': 0.010321324245374878, 'mean_word_length': 2.66869083764958, 'pattern_counts': {'":': 30, '<': 17, '<?xml version=': 0, '>': 30, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 4, 'word_list_counts': {'cursed_substrings.json': 48, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1352892', 'n_tokens_mistral': 6217, 'n_tokens_neox': 5828, 'n_words': 1612}