refactor and cleanup codebase

Author: Arpan-Kreeti

apps/csv2sql/lib/csv2sql.ex

@@ -1,6 +1,6 @@
 defmodule Csv2sql do
  def main(args) do
- # Csv2sql.Helpers.greet()
+ Csv2sql.Helpers.greet()
  # Load configuration varaibles dynamically for escripts, this is required
  # since configuration variables are set to whatever they where when the
  # escript was build and cannot be changed later

apps/csv2sql/lib/csv2sql/application.ex

@@ -2,27 +2,22 @@ defmodule Csv2sql.Application do
  use Application
 
  def start(_type, _args) do
- set_validate = Application.get_env(:csv2sql, Csv2sql.MainServer)[:set_validate]
- set_insert_schema = Application.get_env(:csv2sql, Csv2sql.Worker)[:set_insert_schema]
- set_insert_data = Application.get_env(:csv2sql, Csv2sql.Worker)[:set_insert_data]
-
  repo_supervisor =
- if set_validate || set_insert_schema || set_insert_data do
- [Csv2sql.Repo]
- else
- []
- end
+ if Application.get_env(:csv2sql, Csv2sql.MainServer)[:set_validate] ||
+  Application.get_env(:csv2sql, Csv2sql.Worker)[:set_insert_schema] ||
+  Application.get_env(:csv2sql, Csv2sql.Worker)[:set_insert_data],
+  do: [Csv2sql.Repo],
+  else: []
 
  children =
- []
- |> Kernel.++(repo_supervisor)
- |> Kernel.++([
- Csv2sql.Observer,
- Csv2sql.JobQueueServer,
- Csv2sql.DbWorkerSupervisor,
- Csv2sql.WorkerSupervisor,
- Csv2sql.MainServer
- ])
+ repo_supervisor ++
+ [
+ Csv2sql.Observer,
+ Csv2sql.JobQueueServer,
+ Csv2sql.DbWorkerSupervisor,
+ Csv2sql.WorkerSupervisor,
+ Csv2sql.MainServer
+ ]
 
  opts = [strategy: :one_for_one, name: Csv2sql.Supervisor]
  Supervisor.start_link(children, opts)

apps/csv2sql/lib/csv2sql/data_transfer.ex

@@ -1,8 +1,14 @@
 defmodule Csv2sql.DataTransfer do
  alias NimbleCSV.RFC4180, as: CSV
+ alias Csv2sql.{JobQueueServer, Helpers}
 
+ @doc """
+ Divides a csv file in chunks and place them in a job queue.
+ Whenever a DB worker is free it will pick up a chunk from the queue
+ and insert it in the database.
+ """
  def process_file(file) do
- Csv2sql.Helpers.print_msg("Begin data tranfer for file: " <> Path.basename(file))
+ Helpers.print_msg("Begin data tranfer for file: " <> Path.basename(file))
 
  insertion_chunk_size = Application.get_env(:csv2sql, Csv2sql.Repo)[:insertion_chunk_size]
 
@@ -17,28 +23,35 @@ defmodule Csv2sql.DataTransfer do
  wait_for_file_transfer(file)
  end
 
+
+ # Wait until all chunks for the current file in the job queue has been processed
+ # `:timer.sleep(300)` waits for the last chunk in queue to get inserted that is
+ # if no, chunks were present on the job queue this means a DB worker has picked
+ # up the chunk for insertion, so we wait for 300ms for the chunk to get inserted.
  defp wait_for_file_transfer(file) do
  if Csv2sql.JobQueueServer.job_for_file_present(file) do
  wait_for_file_transfer(file)
  else
  imported_csv_directory =
  Application.get_env(:csv2sql, Csv2sql.MainServer)[:imported_csv_directory]
 
- # wait for the last chunk in queue to get inserted
  :timer.sleep(300)
- File.rename!(file, "#{imported_csv_directory}/#{Path.basename(file)}")
- Csv2sql.Helpers.print_msg("Finished processing file: " <> Path.basename(file), :green)
+ File.rename(file, "#{imported_csv_directory}/#{Path.basename(file)}")
+ Helpers.print_msg("Finished processing file: " <> Path.basename(file), :green)
  end
  end
 
+
+ # Wait until job queue has space for the next chunk
+ # by recursively calling itself.
  defp check_job_queue(file, data_chunk) do
  job_count_limit = Application.get_env(:csv2sql, Csv2sql.Repo)[:job_count_limit]
- job_count = Csv2sql.JobQueueServer.get_job_count()
+ job_count = JobQueueServer.get_job_count()
 
  if job_count > job_count_limit do
  check_job_queue(file, data_chunk)
  else
- Csv2sql.JobQueueServer.add_data_chunk(file, data_chunk)
+ JobQueueServer.add_data_chunk(file, data_chunk)
  end
  end
 end

apps/csv2sql/lib/csv2sql/database.ex

@@ -1,32 +1,26 @@
 defmodule Csv2sql.Database do
  alias NimbleCSV.RFC4180, as: CSV
- alias Csv2sql.Repo
+ alias Csv2sql.{Repo, Helpers, Observer, ErrorTracker}
 
+ @doc """
+ Creates the table for a csv file
+ """
  def make_db_schema([drop_query, create_query]) do
- database_name = Application.get_env(:csv2sql, Csv2sql.Repo)[:database_name]
-
- table_name =
- String.trim_leading(drop_query, "DROP TABLE IF EXISTS #{database_name}.")
- |> String.trim_trailing(";")
-
  try do
  execute_query(drop_query)
  execute_query(create_query)
  catch
  _, reason ->
- Csv2sql.ErrorTracker.add_error(reason)
+ ErrorTracker.add_error(reason)
  end
 
- Csv2sql.Helpers.print_msg("Create Schema for: #{table_name}")
- end
-
- defp execute_query(query) do
- Ecto.Adapters.SQL.query!(Repo, query, [])
+ log_table_created(drop_query)
  end
 
+ @doc """
+ Inserts a chunk of data in the database
+ """
  def insert_data_chunk(file, data_chunk) do
- database_name = Application.get_env(:csv2sql, Csv2sql.Repo)[:database_name]
-
  table_name =
  file
  |> Path.basename()
@@ -46,31 +40,34 @@ defmodule Csv2sql.Database do
  end)
 
  try do
- Repo.insert_all(table_name, data_chunk, prefix: database_name)
+ Repo.insert_all(table_name, data_chunk,
+ prefix: Application.get_env(:csv2sql, Csv2sql.Repo)[:database_name]
+ )
  catch
  _, reason ->
- Csv2sql.ErrorTracker.add_error(reason)
+ ErrorTracker.add_error(reason)
  end
 
- Csv2sql.Observer.update_file_status(file, :insert_data)
+ Observer.update_file_status(file, :insert_data)
  end
 
+ @doc """
+ Prepares Database for data insertion by creating the Database if not exists
+ """
  def prepare_db() do
  database_name = Application.get_env(:csv2sql, Csv2sql.Repo)[:database_name]
 
- Ecto.Adapters.SQL.query!(
- Repo,
- "CREATE DATABASE IF NOT EXISTS #{database_name} CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;",
- []
+ execute_query(
+ "CREATE DATABASE IF NOT EXISTS #{database_name} CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;"
  )
 
- Ecto.Adapters.SQL.query!(
- Repo,
- "SET GLOBAL SQL_MODE=\"NO_BACKSLASH_ESCAPES,NO_ENGINE_SUBSTITUTION,NO_ZERO_IN_DATE\";",
- []
+ execute_query(
+ "SET GLOBAL SQL_MODE=\"NO_BACKSLASH_ESCAPES,NO_ENGINE_SUBSTITUTION,NO_ZERO_IN_DATE\";"
  )
  end
 
+ defp execute_query(query), do: Ecto.Adapters.SQL.query!(Repo, query, [])
+
  defp get_headers(file) do
  [headers] =
  file
@@ -81,4 +78,17 @@ defmodule Csv2sql.Database do
 
  headers
  end
+
+ # Logs the table that was created
+ # Gets the table name from the DROP query
+ defp log_table_created(drop_query) do
+ table_name =
+ drop_query
+ |> String.trim_leading(
+ "DROP TABLE IF EXISTS #{Application.get_env(:csv2sql, Csv2sql.Repo)[:database_name]}."
+ )
+ |> String.trim_trailing(";")
+
+ Helpers.print_msg("Create Schema for: #{table_name}")
+ end
 end

apps/csv2sql/lib/csv2sql/db_worker.ex

@@ -1,7 +1,6 @@
 defmodule Csv2sql.DbWorker do
  use GenServer
-
- alias Csv2sql.JobQueueServer, as: FS
+ alias Csv2sql.{JobQueueServer, Database}
 
  def start_link(_) do
  GenServer.start_link(__MODULE__, :no_args)
@@ -12,11 +11,14 @@ defmodule Csv2sql.DbWorker do
  {:ok, nil}
  end
 
+ @doc """
+ Recursively requests the job queue for work(chunks of data)
+ """
  def handle_info(:start_new_db_work, _) do
- FS.get_work()
+ JobQueueServer.get_work()
  |> case do
  {file, data_chunk} ->
- Csv2sql.Database.insert_data_chunk(file, data_chunk)
+ Database.insert_data_chunk(file, data_chunk)
 
  :no_work ->
  nil

apps/csv2sql/lib/csv2sql/error_tracker.ex

@@ -1,5 +1,6 @@
 defmodule Csv2sql.ErrorTracker do
  use GenServer
+ alias Csv2sql.{Observer, Helpers}
 
  def register_supervisor(sup_pid) do
  GenServer.cast(:error_tracker, {:register_supervisor, sup_pid})
@@ -11,9 +12,9 @@ defmodule Csv2sql.ErrorTracker do
 
  #{inspect(error)}
  """
- |> Csv2sql.Helpers.print_msg(:red)
+ |> Helpers.print_msg(:red)
 
- Csv2sql.Observer.change_stage(:error)
+ Observer.change_stage(:error)
 
  # Call genserver not cast since, we need the wait synchronously untill supervisor is stopped
  GenServer.call(:error_tracker, {:add_error, error})

apps/csv2sql/lib/csv2sql/helpers.ex

@@ -58,16 +58,5 @@ defmodule Csv2sql.Helpers do
  } by Arpan.
  """ <> reset())
  |> IO.puts()
-
- start_text = [:yellow_background, :black, " Starting... !"] |> format()
-
- CliSpinners.spin_fun(
- [
- frames: :arrow3,
- text: start_text,
- done: ""
- ],
- fn -> :timer.sleep(5000) end
- )
  end
 end

apps/csv2sql/lib/csv2sql/import_validator.ex

@@ -2,6 +2,9 @@ defmodule Csv2sql.ImportValidator do
  alias NimbleCSV.RFC4180, as: CSV
  alias Csv2sql.Helpers
 
+ @doc """
+ Validates import by comparing row count in csv vs row count in database
+ """
  def validate_import(file_list) do
  %{stats: {total, correct, incorrect}, incorrect_files: incorrect_files} =
  file_list
@@ -11,19 +14,18 @@ defmodule Csv2sql.ImportValidator do
  %{stats: {total, correct, incorrect}, incorrect_files: incorrect_files} ->
  Helpers.print_msg("Checking File: #{Path.basename(file)}", :yellow)
 
- validated_csv_directory =
- Application.get_env(:csv2sql, Csv2sql.MainServer)[:validated_csv_directory]
+ main_server_config = Application.get_env(:csv2sql, Csv2sql.MainServer)
 
  src =
  if Application.get_env(:csv2sql, Csv2sql.Worker)[:set_insert_data],
- do: Application.get_env(:csv2sql, Csv2sql.MainServer)[:imported_csv_directory],
- else: Application.get_env(:csv2sql, Csv2sql.MainServer)[:source_csv_directory]
+ do: main_server_config[:imported_csv_directory],
+ else: main_server_config[:source_csv_directory]
 
  result =
  if validate_csv(file, row_count) do
  File.rename(
  "#{src}/#{file}.csv",
- "#{validated_csv_directory}/#{file}.csv"
+ "#{main_server_config[:validated_csv_directory]}/#{file}.csv"
  )
 
  %{
@@ -41,36 +43,13 @@ defmodule Csv2sql.ImportValidator do
  end
  )
 
- """
-
- #{IO.ANSI.underline()}Validation Completed:#{IO.ANSI.no_underline()}
-
- """
- |> Helpers.print_msg(:white)
-
- white = IO.ANSI.white()
-
- Helpers.print_msg("* Number of files checked: #{white}#{total}")
- Helpers.print_msg("* Number of files with correct count: #{white}#{correct}")
- Helpers.print_msg("* Number of files with incorrect count: #{white}#{incorrect}")
-
- if incorrect > 0 do
- """
-
-
- Files with incorrect count:
- """
- |> Helpers.print_msg(:white)
-
- incorrect_files
- |> Enum.each(fn file ->
- Helpers.print_msg("* #{Path.basename(file)}", :red)
- end)
- end
-
+ show_validation_results(total, correct, incorrect, incorrect_files)
  {incorrect, incorrect_files}
  end
 
+ @doc """
+ Get row count in csv file
+ """
  def get_count_from_csv(file) do
  file
  |> File.stream!()
@@ -125,4 +104,33 @@ defmodule Csv2sql.ImportValidator do
  -1
  end
  end
+
+ defp show_validation_results(total, correct, incorrect, incorrect_files) do
+ """
+
+ #{IO.ANSI.underline()}Validation Completed:#{IO.ANSI.no_underline()}
+
+ """
+ |> Helpers.print_msg(:white)
+
+ white = IO.ANSI.white()
+
+ Helpers.print_msg("* Number of files checked: #{white}#{total}")
+ Helpers.print_msg("* Number of files with correct count: #{white}#{correct}")
+ Helpers.print_msg("* Number of files with incorrect count: #{white}#{incorrect}")
+
+ if incorrect > 0 do
+ """
+
+
+ Files with incorrect count:
+ """
+ |> Helpers.print_msg(:white)
+
+ incorrect_files
+ |> Enum.each(fn file ->
+ Helpers.print_msg("* #{Path.basename(file)}", :red)
+ end)
+ end
+ end
 end

apps/csv2sql/lib/csv2sql/job_queue_server.ex

@@ -49,11 +49,11 @@ defmodule Csv2sql.JobQueueServer do
  end
 
  def handle_call({:job_for_file_present, file}, _from, state) do
- file_present? =
+ file_present =
  Enum.any?(state, fn {file_job, _data_chunk} ->
  file == file_job
  end)
 
- {:reply, file_present?, state}
+ {:reply, file_present, state}
  end
 end