Add insert loader

src/Loaders/Insert.php

@@ -0,0 +1,118 @@
+<?php
+
+namespace Marquine\Etl\Loaders;
+
+use Generator;
+use Marquine\Etl\Etl;
+
+class Insert implements LoaderInterface
+{
+    /**
+     * The columns to insert.
+     *
+     * @var array
+     */
+    public $columns = [];
+
+    /**
+     * The connection name.
+     *
+     * @var string
+     */
+    public $connection = 'default';
+
+    /**
+     * Indicates if the table has timestamps columns.
+     *
+     * @var bool
+     */
+    public $timestamps = false;
+
+    /**
+     * Transaction mode.
+     *
+     * @var mixed
+     */
+    public $transaction = 'single';
+
+    /**
+     * The insert statement.
+     *
+     * @var \PDOStatement
+     */
+    protected $insert;
+
+    /**
+     * The database table.
+     *
+     * @var string
+     */
+    protected $table;
+
+    /**
+     * Timestamps columns value.
+     *
+     * @var string
+     */
+    protected $time;
+
+    /**
+     * Load data into the given destination.
+     *
+     * @param  string  $destination
+     * @param  \Generator  $data
+     * @return void
+     */
+    public function load(Generator $data, $destination)
+    {
+        $this->normalizeColumns($data);
+
+        $this->table = $destination;
+
+        $this->time = date('Y-m-d G:i:s');
+
+        Etl::database($this->connection)->transaction($this->transaction)->data($data)->run(function ($row) {
+            $this->insert(array_intersect_key($row, $this->columns));
+        });
+    }
+
+    /**
+     * Execute the insert statement.
+     *
+     * @param  array  $row
+     * @return void
+     */
+    protected function insert($row)
+    {
+        if (! $this->insert) {
+            $this->insert = Etl::database($this->connection)->statement()->insert($this->table, $this->columns)->prepare();
+        }
+
+        if ($this->timestamps) {
+            $row['created_at'] = $this->time;
+            $row['updated_at'] = $this->time;
+        }
+
+        $this->insert->execute($row);
+    }
+
+    /**
+     * Normalize columns.
+     *
+     * @param  \Generator  $data
+     * @return void
+     */
+    protected function normalizeColumns($data)
+    {
+        if (empty($this->columns)) {
+            $this->columns = array_keys($data->current());
+        }
+
+        if ($this->timestamps) {
+            $this->columns[] = 'created_at';
+            $this->columns[] = 'updated_at';
+        }
+
+        $this->columns = array_combine($this->columns, $this->columns);
+    }
+}

tests/Loaders/InsertTest.php

@@ -0,0 +1,88 @@
+<?php
+
+namespace Tests\Loaders;
+
+use Tests\TestCase;
+use Marquine\Etl\Etl;
+use Marquine\Etl\Loaders\Insert;
+
+class InsertTest extends TestCase
+{
+    protected function setUp()
+    {
+        parent::setUp();
+
+        $this->createTables();
+    }
+
+    /** @test */
+    function insert_data_into_the_database()
+    {
+        $data = function () {
+            yield ['id' => '1', 'name' => 'Jane Doe', 'email' => 'janedoe@example.com'];
+            yield ['id' => '2', 'name' => 'John Doe', 'email' => 'johndoe@example.com'];
+        };
+
+        $loader = new Insert;
+
+        $loader->load($data(), 'users');
+
+        $expected = [
+            ['id' => '1', 'name' => 'Jane Doe', 'email' => 'janedoe@example.com'],
+            ['id' => '2', 'name' => 'John Doe', 'email' => 'johndoe@example.com'],
+        ];
+
+        $query = Etl::database()->query()->select('users')->execute();
+
+        $this->assertEquals($expected, $query->fetchAll());
+    }
+
+    /** @test */
+    function insert_specified_into_the_database()
+    {
+        $data = function () {
+            yield ['id' => '1', 'name' => 'Jane Doe', 'email' => 'janedoe@example.com'];
+            yield ['id' => '2', 'name' => 'John Doe', 'email' => 'johndoe@example.com'];
+        };
+
+        $loader = new Insert;
+
+        $loader->columns = ['id', 'name'];
+
+        $loader->load($data(), 'users');
+
+
+        $expected = [
+            ['id' => '1', 'name' => 'Jane Doe', 'email' => ''],
+            ['id' => '2', 'name' => 'John Doe', 'email' => ''],
+        ];
+
+        $query = Etl::database()->query()->select('users')->execute();
+
+        $this->assertEquals($expected, $query->fetchAll());
+    }
+
+    /** @test */
+    function insert_data_into_the_database_with_timestamps()
+    {
+        $data = function () {
+            yield ['id' => '1', 'name' => 'Jane Doe', 'email' => 'janedoe@example.com'];
+            yield ['id' => '2', 'name' => 'John Doe', 'email' => 'johndoe@example.com'];
+        };
+
+        $loader = new Insert;
+
+        $loader->timestamps = true;
+
+        $loader->load($data(), 'users_ts');
+
+        $expected = [
+            ['id' => '1', 'name' => 'Jane Doe', 'email' => 'janedoe@example.com', 'created_at' => date('Y-m-d G:i:s'), 'updated_at' => date('Y-m-d G:i:s'), 'deleted_at' => null],
+            ['id' => '2', 'name' => 'John Doe', 'email' => 'johndoe@example.com', 'created_at' => date('Y-m-d G:i:s'), 'updated_at' => date('Y-m-d G:i:s'), 'deleted_at' => null],
+        ];
+
+        $query = Etl::database()->query()->select('users_ts')->execute();
+
+        $this->assertEquals($expected, $query->fetchAll());
+    }
+}