DatabaseSynchronizer::prepareForInserts() - Code Metrics - mtolhuys/laravel-database-synchronizer - Measure and Improve Code Quality continuously with Scrutinizer

DatabaseSynchronizer::prepareForInserts() A
last analyzed 2019-04-08 14:10 UTC

↳ Parent: DatabaseSynchronizer

Complexity

Conditions	5
Paths	7

Size

Total Lines	27
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	16
dl	0
loc	27
rs	9.4222
c	0
b	0
f	0
cc	5
nc	7
nop	1

<?php

namespace mtolhuijs\LDS;

use PDOException;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
use Illuminate\Support\Facades\Artisan;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Database\ConnectionInterface;
use mtolhuijs\LDS\Exceptions\DatabaseConnectionException;

class DatabaseSynchronizer
{
    public const DEFAULT_LIMIT = 5000;

    public $cli;
    public $limit = self::DEFAULT_LIMIT;
    public $tables;
    public $skipTables;
    public $migrate;
    public $from;
    public $to;
    public $truncate;

    private $fromDB;
    private $toDB;

    public function __construct(string $from, string $to, $cli = false)
    {
        $this->from = $from;
        $this->to = $to;
        $this->cli = $cli;

        try {
            $this->fromDB = DB::connection($this->from);
            $this->toDB = DB::connection($this->to);
        } catch (\Exception $e) {
            throw new DatabaseConnectionException($e->getMessage());
        }
    }

    public function run(): void
    {
        if ($this->migrate) {
            Artisan::call('migrate'.($this->truncate ? ':refresh' : ''), [
                '--database' => $this->to,
            ]);
        }

        foreach ($this->getTables() as $table) {
            $this->feedback(PHP_EOL.PHP_EOL."Table: $table", 'line');

            if (! Schema::connection($this->from)->hasTable($table)) {
                $this->feedback("Table '$table' does not exist in $this->from", 'error');

                continue;
            }

            $this->syncTable($table);
            $this->syncRows($table);
        }
    }

    private function createTable(string $table, array $columns): void
    {
        $this->feedback("Creating '$this->to.$table' table", 'warn');

        Schema::connection($this->to)->create($table, function (Blueprint $table_bp) use ($table, $columns) {
            foreach ($columns as $column) {
                $type = Schema::connection($this->from)->getColumnType($table, $column);

                $table_bp->{$type}($column)->nullable();

                $this->feedback("Added {$type}('$column')->nullable()");
            }
        });
    }

    private function updateTable(string $table, string $column): void
    {
        Schema::connection($this->to)->table($table, function (Blueprint $table_bp) use ($table, $column) {
            $type = Schema::connection($this->from)->getColumnType($table, $column);

            $table_bp->{$type}($column)->nullable();

            $this->feedback("Added {$type}('$column')->nullable()");
        });
    }

    public function setSkipTables(array $skipTables)
    {
        $this->skipTables = $skipTables;

        return $this;
    }

    public function setTables(array $tables)
    {
        $this->tables = $tables;

        return $this;
    }

    public function setLimit(int $limit)
    {
        $this->limit = $limit;

        return $this;
    }

    public function setOptions(array $options)
    {
        foreach ($options as $option => $value) {
            if (! isset($this->{$option})) {
                $this->{$option} = $value;
            }
        }

        return $this;
    }

    protected function getFromDb(): ConnectionInterface
    {
        if ($this->fromDB === null) {
            $this->fromDB = DB::connection($this->from);
        }

        return $this->fromDB;
    }

    protected function getToDb(): ConnectionInterface
    {
        if ($this->toDB === null) {
            $this->toDB = DB::connection($this->to);
        }

        return $this->toDB;
    }

    public function getTables(): array
    {
        if (empty($this->tables)) {
            $this->tables = $this->getFromDb()->getDoctrineSchemaManager()->listTableNames();
        }

        return array_filter($this->tables, function ($table) {
            return ! in_array($table, $this->skipTables, true);
        });
    }

    /**
     * Check if tables and columns are present
     * Create or update them if not.
     *
     * @param string $table
     */
    public function syncTable(string $table): void
    {
        $schema = Schema::connection($this->to);
        $columns = Schema::connection($this->from)->getColumnListing($table);

        if ($schema->hasTable($table)) {
            foreach ($columns as $column) {
                if ($schema->hasColumn($table, $column)) {
                    continue;
                }

                $this->updateTable($table, $column);
            }

            return;
        }

        $this->createTable($table, $columns);
    }

    /**
     * Fetch all rows in $this->from and insert or update $this->to.
     * @todo need to get the real primary key
     * @todo add limit offset setup
     * @todo investigate: insert into on duplicate key update
     *
     * @param string $table
     */
    public function syncRows(string $table): void
    {
        $queryColumn = Schema::connection($this->from)->getColumnListing($table)[0];
        $statement = $this->prepareForInserts($table);

        while ($row = $statement->fetch(\PDO::FETCH_OBJ)) {
            $exists = $this->getToDb()->table($table)->where($queryColumn, $row->{$queryColumn})->first();

            if (! $exists) {
                $this->getToDb()->table($table)->insert((array) $row);
            } else {
                $this->getToDb()->table($table)->where($queryColumn, $row->{$queryColumn})->update((array) $row);
            }

            if ($this->cli) {
                $this->cli->progressBar->advance();
            }
        }

        if ($this->cli) {
            $this->cli->progressBar->finish();
        }
    }

    /**
     * @param string $table
     * @return \PDOStatement
     */
    private function prepareForInserts(string $table): \PDOStatement
    {
        $pdo = $this->getFromDb()->getPdo();
        $builder = $this->fromDB->table($table);
        $statement = $pdo->prepare($builder->toSql());

        if (! $statement instanceof \PDOStatement) {
            throw new PDOException("Could not prepare PDOStatement for $table");
        }

        $statement->execute($builder->getBindings());
        $amount = $statement->rowCount();

        if ($this->cli) {
            if ($amount > 0) {
                $this->feedback("Synchronizing '$this->to.$table' rows", 'comment');
                $this->cli->progressBar = $this->cli->getOutput()->createProgressBar($amount);
            } else {
                $this->feedback('No rows...', 'comment');
            }
        }

        if ($this->truncate) {
            $this->getToDb()->table($table)->truncate();
        }

        return $statement;
    }

    private function feedback(string $msg, $type = 'info'): void
    {
        if ($this->cli) {
            $this->cli->{$type}($msg);
        } else {
            echo PHP_EOL.$msg.PHP_EOL;
        }
    }
}


1			<?php
2
3			namespace mtolhuijs\LDS;
4
5			use PDOException;
6			use Illuminate\Support\Facades\DB;
7			use Illuminate\Support\Facades\Schema;
8			use Illuminate\Support\Facades\Artisan;
9			use Illuminate\Database\Schema\Blueprint;
10			use Illuminate\Database\ConnectionInterface;
11			use mtolhuijs\LDS\Exceptions\DatabaseConnectionException;
12
13			class DatabaseSynchronizer
14			{
15			public const DEFAULT_LIMIT = 5000;
16
17			public $cli;
18			public $limit = self::DEFAULT_LIMIT;
19			public $tables;
20			public $skipTables;
21			public $migrate;
22			public $from;
23			public $to;
24			public $truncate;
25
26			private $fromDB;
27			private $toDB;
28
29			public function __construct(string $from, string $to, $cli = false)
30			{
31			$this->from = $from;
32			$this->to = $to;
33			$this->cli = $cli;
34
35			try {
36			$this->fromDB = DB::connection($this->from);
37			$this->toDB = DB::connection($this->to);
38			} catch (\Exception $e) {
39			throw new DatabaseConnectionException($e->getMessage());
40			}
41			}
42
43			public function run(): void
44			{
45			if ($this->migrate) {
46			Artisan::call('migrate'.($this->truncate ? ':refresh' : ''), [
47			'--database' => $this->to,
48			]);
49			}
50
51			foreach ($this->getTables() as $table) {
52			$this->feedback(PHP_EOL.PHP_EOL."Table: $table", 'line');
53
54			if (! Schema::connection($this->from)->hasTable($table)) {
55			$this->feedback("Table '$table' does not exist in $this->from", 'error');
56
57			continue;
58			}
59
60			$this->syncTable($table);
61			$this->syncRows($table);
62			}
63			}
64
65			private function createTable(string $table, array $columns): void
66			{
67			$this->feedback("Creating '$this->to.$table' table", 'warn');
68
69			Schema::connection($this->to)->create($table, function (Blueprint $table_bp) use ($table, $columns) {
70			foreach ($columns as $column) {
71			$type = Schema::connection($this->from)->getColumnType($table, $column);
72
73			$table_bp->{$type}($column)->nullable();
74
75			$this->feedback("Added {$type}('$column')->nullable()");
76			}
77			});
78			}
79
80			private function updateTable(string $table, string $column): void
81			{
82			Schema::connection($this->to)->table($table, function (Blueprint $table_bp) use ($table, $column) {
83			$type = Schema::connection($this->from)->getColumnType($table, $column);
84
85			$table_bp->{$type}($column)->nullable();
86
87			$this->feedback("Added {$type}('$column')->nullable()");
88			});
89			}
90
91			public function setSkipTables(array $skipTables)
92			{
93			$this->skipTables = $skipTables;
94
95			return $this;
96			}
97
98			public function setTables(array $tables)
99			{
100			$this->tables = $tables;
101
102			return $this;
103			}
104
105			public function setLimit(int $limit)
106			{
107			$this->limit = $limit;
108
109			return $this;
110			}
111
112			public function setOptions(array $options)
113			{
114			foreach ($options as $option => $value) {
115			if (! isset($this->{$option})) {
116			$this->{$option} = $value;
117			}
118			}
119
120			return $this;
121			}
122
123			protected function getFromDb(): ConnectionInterface
124			{
125			if ($this->fromDB === null) {
126			$this->fromDB = DB::connection($this->from);
127			}
128
129			return $this->fromDB;
130			}
131
132			protected function getToDb(): ConnectionInterface
133			{
134			if ($this->toDB === null) {
135			$this->toDB = DB::connection($this->to);
136			}
137
138			return $this->toDB;
139			}
140
141			public function getTables(): array
142			{
143			if (empty($this->tables)) {
144			$this->tables = $this->getFromDb()->getDoctrineSchemaManager()->listTableNames();
145			}
146
147			return array_filter($this->tables, function ($table) {
148			return ! in_array($table, $this->skipTables, true);
149			});
150			}
151
152			/**
153			* Check if tables and columns are present
154			* Create or update them if not.
155			*
156			* @param string $table
157			*/
158			public function syncTable(string $table): void
159			{
160			$schema = Schema::connection($this->to);
161			$columns = Schema::connection($this->from)->getColumnListing($table);
162
163			if ($schema->hasTable($table)) {
164			foreach ($columns as $column) {
165			if ($schema->hasColumn($table, $column)) {
166			continue;
167			}
168
169			$this->updateTable($table, $column);
170			}
171
172			return;
173			}
174
175			$this->createTable($table, $columns);
176			}
177
178			/**
179			* Fetch all rows in $this->from and insert or update $this->to.
180			* @todo need to get the real primary key
181			* @todo add limit offset setup
182			* @todo investigate: insert into on duplicate key update
183			*
184			* @param string $table
185			*/
186			public function syncRows(string $table): void
187			{
188			$queryColumn = Schema::connection($this->from)->getColumnListing($table)[0];
189			$statement = $this->prepareForInserts($table);
190
191			while ($row = $statement->fetch(\PDO::FETCH_OBJ)) {
192			$exists = $this->getToDb()->table($table)->where($queryColumn, $row->{$queryColumn})->first();
193
194			if (! $exists) {
195			$this->getToDb()->table($table)->insert((array) $row);
196			} else {
197			$this->getToDb()->table($table)->where($queryColumn, $row->{$queryColumn})->update((array) $row);
198			}
199
200			if ($this->cli) {
201			$this->cli->progressBar->advance();
202			}
203			}
204
205			if ($this->cli) {
206			$this->cli->progressBar->finish();
207			}
208			}
209
210			/**
211			* @param string $table
212			* @return \PDOStatement
213			*/
214			private function prepareForInserts(string $table): \PDOStatement
215			{
216			$pdo = $this->getFromDb()->getPdo();
217			$builder = $this->fromDB->table($table);
218			$statement = $pdo->prepare($builder->toSql());
219
220			if (! $statement instanceof \PDOStatement) {
221			throw new PDOException("Could not prepare PDOStatement for $table");
222			}
223
224			$statement->execute($builder->getBindings());
225			$amount = $statement->rowCount();
226
227			if ($this->cli) {
228			if ($amount > 0) {
229			$this->feedback("Synchronizing '$this->to.$table' rows", 'comment');
230			$this->cli->progressBar = $this->cli->getOutput()->createProgressBar($amount);
231			} else {
232			$this->feedback('No rows...', 'comment');
233			}
234			}
235
236			if ($this->truncate) {
237			$this->getToDb()->table($table)->truncate();
238			}
239
240			return $statement;
241			}
242
243			private function feedback(string $msg, $type = 'info'): void
244			{
245			if ($this->cli) {
246			$this->cli->{$type}($msg);
247			} else {
248			echo PHP_EOL.$msg.PHP_EOL;
249			}
250			}
251			}
252

mtolhuys / laravel-database-synchronizer

DatabaseSynchronizer::prepareForInserts() A last analyzed 2019-04-08 14:10 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

DatabaseSynchronizer::prepareForInserts() A
last analyzed 2019-04-08 14:10 UTC