Extraction

Extract

EXTRACT starts by securely connecting to the specified SQL database using encrypted credentials stored with keyring. The combination of configuration and SQL scripts determine which data will be extracted by filtering based on programme(s) of study and specifying which nodes, relationships and properties to extract. Additional options include specifying chunk size if extracting significant amounts of data, for example.

The process performs basic validation at every step ensuring secure connection before running SQL SELECT statements and storing extracted data as local csv files.

SQL example

SELECT DISTINCT a.[Id] AS actSplusID,
     CONCAT(a.[Id], '-', adt.[Week], '-', adt.[Day]) AS actGraphID,
     a.[Name] AS actName,
     a.[Description] AS actDescription,
     a.[DepartmentId] AS actDeptSPlusID,
     adt.[StartDateTime] AS actStartDateTime,
     adt.[EndDateTime] AS actEndDateTime,
     adt.[Week] AS actWeekNum,
     adt.[Occurrence] AS actOccurrence,
     a.[ModuleId] AS actModSplusID,
     a.[ScheduledDay] AS actScheduledDay,
     a.[StartDate] AS actFirstActivityDate,
     a.[EndDate] AS actLastActivityDate,
     a.[PlannedSize] AS actPlannedSize,
     a.[RealSize] AS actRealSize,
     a.[Duration] AS actDuration,
     a.[DurationInMinutes] AS actDurationInMinutes,
     a.[NumberOfOccurrences] AS actNumberOfOccurrences,
     a.[WeekPattern] AS actWeekPattern,
     a.[ActivityTypeId] AS actActivityTypeSplusID,
     a.[WhenScheduled] AS actWhenScheduled,
     a.[IsJtaParent],
     a.[IsJtaChild],
     a.[IsVariantParent],
     a.[IsVariantChild]
FROM ##TempActivity a
INNER JOIN ##TempActivityDateTime adt ON a.[Id] = adt.[ActivityID];

Snippet: extract_data.py

Click to show code

# extract_main.py
from logger_config import extract_logger
from extract_data import main as extract_main
from config import EXTRACT_DIR, HOSTKEYS, CHUNK_SIZE
from utils import execution_times

def run_extraction():
    extract_logger.info("Starting data extraction process")
    extract_logger.info(f"Output Directory: {EXTRACT_DIR}")
    extract_logger.info(f"Hostkeys: {HOSTKEYS}")
    extract_logger.info(f"Chunksize: {CHUNK_SIZE}")

    try:
        extract_main()
    except Exception as e:
        extract_logger.exception("An error occurred during data extraction:")
    finally:
        extract_logger.info("Data extraction completed.")

   
    # Log the execution times
    extract_logger.info("Extraction Time Summary:")
    for func_name, exec_time in execution_times.items():
        extract_logger.info(f"Function {func_name} took {exec_time:.2f} seconds")


if __name__ == "__main__":
    run_extraction()