ElasticConnection::createType() - Code Metrics - Inspection of "Feature: bulk operations - ElasticConnection" - andsalves/doctrine-elastic - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( a53f28...120c3a )

by Anderson

created 2017-09-10 23:22 UTC

ElasticConnection::createType() B

↳ Parent: ElasticConnection

Complexity

Conditions	4
Paths	4

Size

Total Lines	26
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	26
rs	8.5806
c	0
b	0
f	0
cc	4
eloc	14
nc	4
nop	4

<?php

namespace DoctrineElastic\Connection;

use DoctrineElastic\Exception\ConnectionException;
use DoctrineElastic\Exception\ElasticOperationException;
use DoctrineElastic\Exception\InvalidParamsException;
use DoctrineElastic\Helper\MappingHelper;
use DoctrineElastic\Http\CurlRequest;
use DoctrineElastic\Traiting\ErrorGetterTrait;

/**
 * Default elastic connection class for general operations
 * Notice that the original elastic result of most of operations can be get by $return param
 *
 * @author Andsalves <[email protected]>
 */
class ElasticConnection implements ElasticConnectionInterface {

    use ErrorGetterTrait;

    /** Override default elastic limit size query */
    const DEFAULT_MAX_RESULTS = 10000;

    /** @var CurlRequest */
    protected $curlRequest;

    /** @var float */
    protected $esVersion;

    public function __construct(array $hosts) {
        $this->curlRequest = new CurlRequest();
        $baseHost = reset($hosts);

        if (empty($baseHost) || !is_string($baseHost) || !preg_match('/http/', $baseHost)) {
            throw new ConnectionException("Elasticsearch host is invalid. ");
        }

        $this->curlRequest->setBaseUrl($baseHost);
    }

    /**
     * @param string $index
     * @param array|null $mappings
     * @param array|null $settings
     * @param array|null $aliases
     * @param array|null $return
     * @return bool
     */
    public function createIndex(
        $index, array $mappings = null, array $settings = null, array $aliases = null, array &$return = null
    ) {
        if ($this->indexExists($index)) {
            throw new \InvalidArgumentException(sprintf("'%s' index already exists", $index));
        }

        $params = [];

        if (is_array($mappings) && !empty($mappings)) {
            $params['mappings'] = MappingHelper::patchMappings($mappings, floor($this->getElasticsearchVersion()));
        }

        if (is_array($settings) && !empty($settings)) {
            $params['settings'] = $settings;
        }

        if (is_array($aliases) && !empty($aliases)) {
            $params['aliases'] = $aliases;
        }

        $response = $this->curlRequest->request($index, $params, 'PUT');
        $return = $response['content'];

        if (isset($return['acknowledged']) && $return['acknowledged']) {
            return $return['acknowledged'];
        }

        $this->setErrorFromElasticReturn($return);

        return false;
    }

    /**
     * @param string $index
     * @param array|null $return
     * @return bool
     * @throws ElasticOperationException
     */
    public function deleteIndex($index, array &$return = null) {
        if (is_string($index) && !strstr('_all', $index) && !strstr('*', $index)) {
            $response = $this->curlRequest->request("$index?refresh=true", [], 'DELETE');
            $return = $response['content'];

            if ($response['status'] == 404) {
                throw new ElasticOperationException("Index '$index' doesn't exist so cannot be deleted. ");
            }

            if (isset($return['acknowledged'])) {
                return $return['acknowledged'];
            }
        } else {
            throw new ElasticOperationException('Index name is invalid for deletion. ');
        }

        $this->setErrorFromElasticReturn($return);

        return false;
    }

    /**
     * @param string $index
     * @param string $type
     * @param array $mappings
     * @param array|null $return
     * @return bool
     * @throws ElasticOperationException
     */
    public function createType($index, $type, array $mappings = [], array &$return = null) {
        if (!$this->indexExists($index)) {
            throw new \InvalidArgumentException(sprintf("%s' index does not exists", $index));
        }

        if ($this->typeExists($index, $type)) {
            throw new \InvalidArgumentException(sprintf("Type 's%' already exists on index %s", $type, $index));
        }

        $mappings = MappingHelper::patchMappings($mappings, floor($this->getElasticsearchVersion()));

        $url = "$index/_mapping/$type";
        $response = $this->curlRequest->request($url, $mappings, 'PUT');

        $this->throwExceptionFromResponse($response, "Error creating type '$type' in '$index' index");

        $return = $response['content'];

        if (isset($return['acknowledged'])) {
            return $return['acknowledged'];
        }

        $this->setErrorFromElasticReturn($return);

        return false;
    }

    /**
     * @param string $index
     * @param string $type
     * @param array $body
     * @param array $queryParams
     * @param array|null $return
     * @return bool
     */
    public function insert($index, $type, array $body, array $queryParams = [], array &$return = null) {
        $url = "$index/$type";
        if (isset($body['_id'])) {
            $url .= '/' . $body['_id'];
            unset($body['_id']);
        }

        $url = "$url?" . http_build_query(array_merge(['refresh' => "true"], $queryParams));

        $response = $this->curlRequest->request($url, $body, 'POST');

        $this->throwExceptionFromResponse($response);
        $return = $response['content'];

        if (isset($return['created'])) {
            return $return['created'];
        }

        $this->setErrorFromElasticReturn($return);

        return false;
    }

    /**
     * @param string $index
     * @param string $type
     * @param string $_id
     * @param array $body
     * @param array $queryParams
     * @param array|null $return
     *
     * @return bool
     */
    public function update($index, $type, $_id, array $body = [], array $queryParams = [], array &$return = null) {
        if (!$this->indexExists($index)) {
            return false;
        }

        if (array_key_exists('doc', $body)) {
            $params = $body;
        } else {
            $params = ['doc' => $body];
        }

        $url = "$index/$type/$_id/_update?" . http_build_query(array_merge(['refresh' => 'true'], $queryParams));
        $response = $this->curlRequest->request($url, $params, 'POST');
        $this->throwExceptionFromResponse($response);

        $return = $response['content'];

        if (isset($return['_id'])) {
            return true;
        }

        $this->setErrorFromElasticReturn($return);

        return false;
    }

    /**
     * @param string $index
     * @param string $type
     * @param string $_id
     * @param array $queryParams
     * @param array|null $return
     * @return bool
     */
    public function delete($index, $type, $_id, array $queryParams = [], array &$return = null) {
        if (!$this->indexExists($index)) {
            return false;
        }

        $url = "$index/$type/$_id?" . http_build_query(array_merge(['refresh' => 'true'], $queryParams));
        $response = $this->curlRequest->request($url, [], 'DELETE');
        $this->throwExceptionFromResponse($response);
        $return = $response['content'];

        if (isset($return['found']) && !$return['found']) {
            error_log("Doc with _id '$_id' was not found for delete. Index: '$index', Type: '$type' ");
        }

        if (isset($return['_id'])) {
            return true;
        }

        $this->setErrorFromElasticReturn($return);

        return false;
    }

    public function updateWhere($index, $type, array $where, array &$return = null) {
        // TODO
    }

    public function deleteWhere($index, $type, array $where, array &$return = null) {
        // TODO
    }

    /**
     *
     * @param string $index
     * @param string $type
     * @param string $_id
     * @param array $queryParams
     * @param array|null $return
     * @return array|null
     */
    public function get($index, $type, $_id, array $queryParams = [], array &$return = null) {
        if (!$this->indexExists($index)) {
            return null;
        }

        $url = "$index/$type/$_id";
        $response = $this->curlRequest->request($url, [], 'GET');
        $return = $response['content'];

        if ($response['status'] == 404) {
            return null;
        }

        if (isset($return['found']) && boolval($return['found'])) {
            return $return;
class Author {
    private $name;

    public function __construct($name) {
        $this->name = $name;
    }

    public function getName() {
        return $this->name;
    }
}

abstract class Post {
    public function getAuthor() {
        return 'Johannes';
    }
}

class BlogPost extends Post {
    public function getAuthor() {
        return new Author('Johannes');
    }
}

class ForumPost extends Post { /* ... */ }

function my_function(Post $post) {
    echo strtoupper($post->getAuthor());
}
        }

        return null;
    }

    /**
     * Returns the [hits][hits] array from query
     *
     * @param string $index
     * @param string $type
     * @param array $body
     * @param array $queryParams
     * @param array|null $return
     * @return array
     */
    public function search($index, $type, array $body = [], array $queryParams = [], array &$return = null) {
        if (!$this->indexExists($index)) {
            return [];
        }

        $body = $this->unsetEmpties($body);

        if (isset($body['query']) && empty($body['query'])) {
            unset($body['query']);
        }

        $url = "$index/$type/_search?" . http_build_query($queryParams);

        $cleanQuery = function ($queryPart, callable $recusiveFn) {
            if (!is_array($queryPart)) {
                return $queryPart;
            }

            foreach ($queryPart as $key => $item) {
                if ($key == 'query' && isset($queryPart['query']['bool'])) {
                    $queryPart['bool'] = $recusiveFn($queryPart['query']['bool'], $recusiveFn);
                    unset($queryPart['query']['bool']);
                }

                if (isset($item['query']['bool'])) {
                    $queryPart[$key]['bool'] = $recusiveFn($item['query']['bool'], $recusiveFn);
                    unset($queryPart[$key]['query']);
                } else if (is_array($item)) {
                    $queryPart[$key] = $recusiveFn($item, $recusiveFn);
                }
            }

            return $queryPart;
        };

        if (isset($body['query'])) {
            foreach ($body['query']['bool'] as $key => $item) {
                $body['query']['bool'][$key] = $cleanQuery($item, $cleanQuery);
            }
        }

        $response = $this->curlRequest->request($url, $body, 'POST');
        $this->throwExceptionFromResponse($response);
        $return = $response['content'];

        if (isset($return['hits']['hits'])) {
            return $return['hits']['hits'];
        }

        return [];
    }

    /**
     * @param $action
     * @param $index
     * @param $type
     * @param array|null $data
     * @return array|bool
     * @throws InvalidParamsException
     */
    public function bulk($action, $index, $type, array $data = null) {
        $bulkData = '';

        if (!in_array($action, ['create', 'index', 'update']) && !is_array($data)) {
            throw new InvalidParamsException(
                "\$data param must be an array for '$action' bulk action"
            );
        }

        switch ($action) {
            case 'create':
            case 'index':
                foreach ($data as $doc) {
$collection = json_decode($data, true);
if ( ! is_array($collection)) {
    throw new \RuntimeException('$collection must be an array.');
}

foreach ($collection as $item) { /** ... */ }
                    $actionParams = ['_index' => $index, '_type' => $type];

                    if (isset($doc['_id'])) {
                        $actionParams['_id'] = $doc['_id'];
                        unset($doc['_id']);
                    }

                    $bulkData .= json_encode([$action => $actionParams]) . "\n";
                    $bulkData .= json_encode($doc) . "\n";
                }
                break;
            case 'update':
            case 'delete':
                foreach ($data as $doc) {
$collection = json_decode($data, true);
if ( ! is_array($collection)) {
    throw new \RuntimeException('$collection must be an array.');
}

foreach ($collection as $item) { /** ... */ }
                    if (!isset($doc['_id'])) {
                        throw new InvalidParamsException(
                            "_id field must be provided for each item in \$data param for $action action"
                        );
                    }

                    $actionParams = ['_index' => $index, '_type' => $type, '_id' => $doc['_id']];
                    unset($doc['_id']);

                    if ($action == 'update') {
                        if (!array_key_exists('doc', $doc)) {
                            $doc = ['doc' => $doc];
                        }

                        $bulkData .= json_encode([$action => $actionParams]) . "\n";
                        $bulkData .= json_encode($doc) . "\n";
                    }
                }

                break;
            default:
                $bulkActions = ['index', 'create', 'delete', 'update'];

                throw new InvalidParamsException(
                    "Invalid 'action' param provided. Must be one of those: " . implode('|', $bulkActions)
                );
        }

        $response = $this->curlRequest->request("/_bulk", $bulkData, 'POST', [
            'Content-Type: application/x-ndjson; charset=utf-8'
        ]);

        if (isset($response['status']) && $response['status'] == 200) {
            return $response['content'];
        } elseif (isset($response['content'])) {
            return $response['content'];
        }

        return false;
    }

    private function unsetEmpties(array $haystack) {
        $selfFn = __FUNCTION__;

        foreach ($haystack as $key => $value) {
            if (is_array($value)) {
                $haystack[$key] = $this->$selfFn($haystack[$key]);
            }

            if (empty($haystack[$key])) {
                unset($haystack[$key]);
            }
        }

        return $haystack;
    }

    /**
     * @param string $index
     * @return bool
     */
    public function indexExists($index) {
        $response = $this->curlRequest->request($index, [], 'HEAD');

        return $response['status'] === 200;
    }

    /**
     * @param string $index
     * @param string $type
     * @return bool
     */
    public function typeExists($index, $type) {
        $response = $this->curlRequest->request("$index/$type", [], 'HEAD');

        return $response['status'] === 200;
    }

    private function throwExceptionFromResponse($response, $appendPrefix = '') {
        if (isset($response['content']['error']['reason'])) {
            if (!empty($appendPrefix)) {
                $appendPrefix .= ': ';
            }

            throw new ElasticOperationException($appendPrefix . $response['content']['error']['reason']);
        }
    }

    public function hasConnection() {
        $response = $this->curlRequest->request('', [], 'HEAD');

        return $response['status'] == 200;
    }

    private function setErrorFromElasticReturn($return) {
        if (isset($return['error']['root_cause'][0]['reason'])) {
            $this->setError($return['error']['root_cause'][0]['reason']);
        } else if (isset($return['error']['reason'])) {
            $this->setError($return['error']['reason']);
        }
    }

    public function getElasticsearchVersion() {
        if (is_null($this->esVersion)) {
            $response = $this->curlRequest->request('', [], 'GET');

            if (isset($response['content']['version']['number'])) {
                $this->esVersion = floatval($response['content']['version']['number']);
            } else {
                throw new ConnectionException('Unable to fetch elasticsearch version. ');
            }
        }

        return $this->esVersion;
    }
}


1			<?php
2
3			namespace DoctrineElastic\Connection;
4
5			use DoctrineElastic\Exception\ConnectionException;
6			use DoctrineElastic\Exception\ElasticOperationException;
7			use DoctrineElastic\Exception\InvalidParamsException;
8			use DoctrineElastic\Helper\MappingHelper;
9			use DoctrineElastic\Http\CurlRequest;
10			use DoctrineElastic\Traiting\ErrorGetterTrait;
11
12			/**
13			* Default elastic connection class for general operations
14			* Notice that the original elastic result of most of operations can be get by $return param
15			*
16			* @author Andsalves <[email protected]>
17			*/
18			class ElasticConnection implements ElasticConnectionInterface {
19
20			use ErrorGetterTrait;
21
22			/** Override default elastic limit size query */
23			const DEFAULT_MAX_RESULTS = 10000;
24
25			/** @var CurlRequest */
26			protected $curlRequest;
27
28			/** @var float */
29			protected $esVersion;
30
31			public function __construct(array $hosts) {
32			$this->curlRequest = new CurlRequest();
33			$baseHost = reset($hosts);
34
35			if (empty($baseHost) \|\| !is_string($baseHost) \|\| !preg_match('/http/', $baseHost)) {
36			throw new ConnectionException("Elasticsearch host is invalid. ");
37			}
38
39			$this->curlRequest->setBaseUrl($baseHost);
40			}
41
42			/**
43			* @param string $index
44			* @param array\|null $mappings
45			* @param array\|null $settings
46			* @param array\|null $aliases
47			* @param array\|null $return
48			* @return bool
49			*/
50			public function createIndex(
51			$index, array $mappings = null, array $settings = null, array $aliases = null, array &$return = null
52			) {
53			if ($this->indexExists($index)) {
54			throw new \InvalidArgumentException(sprintf("'%s' index already exists", $index));
55			}
56
57			$params = [];
58
59			if (is_array($mappings) && !empty($mappings)) {
60			$params['mappings'] = MappingHelper::patchMappings($mappings, floor($this->getElasticsearchVersion()));
61			}
62
63			if (is_array($settings) && !empty($settings)) {
64			$params['settings'] = $settings;
65			}
66
67			if (is_array($aliases) && !empty($aliases)) {
68			$params['aliases'] = $aliases;
69			}
70
71			$response = $this->curlRequest->request($index, $params, 'PUT');
72			$return = $response['content'];
73
74			if (isset($return['acknowledged']) && $return['acknowledged']) {
75			return $return['acknowledged'];
76			}
77
78			$this->setErrorFromElasticReturn($return);
79
80			return false;
81			}
82
83			/**
84			* @param string $index
85			* @param array\|null $return
86			* @return bool
87			* @throws ElasticOperationException
88			*/
89			public function deleteIndex($index, array &$return = null) {
90			if (is_string($index) && !strstr('_all', $index) && !strstr('*', $index)) {
91			$response = $this->curlRequest->request("$index?refresh=true", [], 'DELETE');
92			$return = $response['content'];
93
94			if ($response['status'] == 404) {
95			throw new ElasticOperationException("Index '$index' doesn't exist so cannot be deleted. ");
96			}
97
98			if (isset($return['acknowledged'])) {
99			return $return['acknowledged'];
100			}
101			} else {
102			throw new ElasticOperationException('Index name is invalid for deletion. ');
103			}
104
105			$this->setErrorFromElasticReturn($return);
106
107			return false;
108			}
109
110			/**
111			* @param string $index
112			* @param string $type
113			* @param array $mappings
114			* @param array\|null $return
115			* @return bool
116			* @throws ElasticOperationException
117			*/
118			public function createType($index, $type, array $mappings = [], array &$return = null) {
119			if (!$this->indexExists($index)) {
120			throw new \InvalidArgumentException(sprintf("%s' index does not exists", $index));
121			}
122
123			if ($this->typeExists($index, $type)) {
124			throw new \InvalidArgumentException(sprintf("Type 's%' already exists on index %s", $type, $index));
125			}
126
127			$mappings = MappingHelper::patchMappings($mappings, floor($this->getElasticsearchVersion()));
128
129			$url = "$index/_mapping/$type";
130			$response = $this->curlRequest->request($url, $mappings, 'PUT');
131
132			$this->throwExceptionFromResponse($response, "Error creating type '$type' in '$index' index");
133
134			$return = $response['content'];
135
136			if (isset($return['acknowledged'])) {
137			return $return['acknowledged'];
138			}
139
140			$this->setErrorFromElasticReturn($return);
141
142			return false;
143			}
144
145			/**
146			* @param string $index
147			* @param string $type
148			* @param array $body
149			* @param array $queryParams
150			* @param array\|null $return
151			* @return bool
152			*/
153			public function insert($index, $type, array $body, array $queryParams = [], array &$return = null) {
154			$url = "$index/$type";
155			if (isset($body['_id'])) {
156			$url .= '/' . $body['_id'];
157			unset($body['_id']);
158			}
159
160			$url = "$url?" . http_build_query(array_merge(['refresh' => "true"], $queryParams));
161
162			$response = $this->curlRequest->request($url, $body, 'POST');
163
164			$this->throwExceptionFromResponse($response);
165			$return = $response['content'];
166
167			if (isset($return['created'])) {
168			return $return['created'];
169			}
170
171			$this->setErrorFromElasticReturn($return);
172
173			return false;
174			}
175
176			/**
177			* @param string $index
178			* @param string $type
179			* @param string $_id
180			* @param array $body
181			* @param array $queryParams
182			* @param array\|null $return
183			*
184			* @return bool
185			*/
186			public function update($index, $type, $_id, array $body = [], array $queryParams = [], array &$return = null) {
187			if (!$this->indexExists($index)) {
188			return false;
189			}
190
191			if (array_key_exists('doc', $body)) {
192			$params = $body;
193			} else {
194			$params = ['doc' => $body];
195			}
196
197			$url = "$index/$type/$_id/_update?" . http_build_query(array_merge(['refresh' => 'true'], $queryParams));
198			$response = $this->curlRequest->request($url, $params, 'POST');
199			$this->throwExceptionFromResponse($response);
200
201			$return = $response['content'];
202
203			if (isset($return['_id'])) {
204			return true;
205			}
206
207			$this->setErrorFromElasticReturn($return);
208
209			return false;
210			}
211
212			/**
213			* @param string $index
214			* @param string $type
215			* @param string $_id
216			* @param array $queryParams
217			* @param array\|null $return
218			* @return bool
219			*/
220			public function delete($index, $type, $_id, array $queryParams = [], array &$return = null) {
221			if (!$this->indexExists($index)) {
222			return false;
223			}
224
225			$url = "$index/$type/$_id?" . http_build_query(array_merge(['refresh' => 'true'], $queryParams));
226			$response = $this->curlRequest->request($url, [], 'DELETE');
227			$this->throwExceptionFromResponse($response);
228			$return = $response['content'];
229
230			if (isset($return['found']) && !$return['found']) {
231			error_log("Doc with _id '$_id' was not found for delete. Index: '$index', Type: '$type' ");
232			}
233
234			if (isset($return['_id'])) {
235			return true;
236			}
237
238			$this->setErrorFromElasticReturn($return);
239
240			return false;
241			}
242
243			public function updateWhere($index, $type, array $where, array &$return = null) {
244			// TODO
245			}
246
247			public function deleteWhere($index, $type, array $where, array &$return = null) {
248			// TODO
249			}
250
251			/**
252			*
253			* @param string $index
254			* @param string $type
255			* @param string $_id
256			* @param array $queryParams
257			* @param array\|null $return
258			* @return array\|null
259			*/
260			public function get($index, $type, $_id, array $queryParams = [], array &$return = null) {
261			if (!$this->indexExists($index)) {
262			return null;
263			}
264
265			$url = "$index/$type/$_id";
266			$response = $this->curlRequest->request($url, [], 'GET');
267			$return = $response['content'];
268
269			if ($response['status'] == 404) {
270			return null;
271			}
272
273			if (isset($return['found']) && boolval($return['found'])) {
274			return $return;
			0 ignored issues – show Bug Best Practice introduced 2017-05-29 00:12 UTC by Report Bug Copy Issue Report The return type of `return $return;` (`object\|integer\|double\|string\|array\|boolean`) is incompatible with the return type declared by the interface `DoctrineElastic\Connecti...onnectionInterface::get` of type `array\|null`. If you return a value from a function or method, it should be a sub-type of the type that is given by the parent type f.e. an interface, or abstract method. This is more formally defined by the Lizkov substitution principle, and guarantees that classes that depend on the parent type can use any instance of a child type interchangably. This principle also belongs to the SOLID principles for object oriented design. Let’s take a look at an example: class Author { private $name; public function __construct($name) { $this->name = $name; } public function getName() { return $this->name; } } abstract class Post { public function getAuthor() { return 'Johannes'; } } class BlogPost extends Post { public function getAuthor() { return new Author('Johannes'); } } class ForumPost extends Post { /* ... */ } function my_function(Post $post) { echo strtoupper($post->getAuthor()); } Our function `my_function` expects a `Post` object, and outputs the author of the post. The base class `Post` returns a simple string and outputting a simple string will work just fine. However, the child class `BlogPost` which is a sub-type of `Post` instead decided to return an `object`, and is therefore violating the SOLID principles. If a `BlogPost` were passed to `my_function`, PHP would not complain, but ultimately fail when executing the `strtoupper` call in its body. Loading history...
275			}
276
277			return null;
278			}
279
280			/**
281			* Returns the [hits][hits] array from query
282			*
283			* @param string $index
284			* @param string $type
285			* @param array $body
286			* @param array $queryParams
287			* @param array\|null $return
288			* @return array
289			*/
290			public function search($index, $type, array $body = [], array $queryParams = [], array &$return = null) {
291			if (!$this->indexExists($index)) {
292			return [];
293			}
294
295			$body = $this->unsetEmpties($body);
296
297			if (isset($body['query']) && empty($body['query'])) {
298			unset($body['query']);
299			}
300
301			$url = "$index/$type/_search?" . http_build_query($queryParams);
302
303			$cleanQuery = function ($queryPart, callable $recusiveFn) {
304			if (!is_array($queryPart)) {
305			return $queryPart;
306			}
307
308			foreach ($queryPart as $key => $item) {
309			if ($key == 'query' && isset($queryPart['query']['bool'])) {
310			$queryPart['bool'] = $recusiveFn($queryPart['query']['bool'], $recusiveFn);
311			unset($queryPart['query']['bool']);
312			}
313
314			if (isset($item['query']['bool'])) {
315			$queryPart[$key]['bool'] = $recusiveFn($item['query']['bool'], $recusiveFn);
316			unset($queryPart[$key]['query']);
317			} else if (is_array($item)) {
318			$queryPart[$key] = $recusiveFn($item, $recusiveFn);
319			}
320			}
321
322			return $queryPart;
323			};
324
325			if (isset($body['query'])) {
326			foreach ($body['query']['bool'] as $key => $item) {
327			$body['query']['bool'][$key] = $cleanQuery($item, $cleanQuery);
328			}
329			}
330
331			$response = $this->curlRequest->request($url, $body, 'POST');
332			$this->throwExceptionFromResponse($response);
333			$return = $response['content'];
334
335			if (isset($return['hits']['hits'])) {
336			return $return['hits']['hits'];
337			}
338
339			return [];
340			}
341
342			/**
343			* @param $action
344			* @param $index
345			* @param $type
346			* @param array\|null $data
347			* @return array\|bool
348			* @throws InvalidParamsException
349			*/
350			public function bulk($action, $index, $type, array $data = null) {
351			$bulkData = '';
352
353			if (!in_array($action, ['create', 'index', 'update']) && !is_array($data)) {
354			throw new InvalidParamsException(
355			"\$data param must be an array for '$action' bulk action"
356			);
357			}
358
359			switch ($action) {
360			case 'create':
361			case 'index':
362			foreach ($data as $doc) {
			0 ignored issues – show Bug introduced 2017-09-10 23:30 UTC by Report Bug Copy Issue Report The expression `$data` of type `null\|array` is not guaranteed to be traversable. How about adding an additional type check? There are different options of fixing this problem. If you want to be on the safe side, you can add an additional type-check: $collection = json_decode($data, true); if ( ! is_array($collection)) { throw new \RuntimeException('$collection must be an array.'); } foreach ($collection as $item) { /** ... / } If you are sure that the expression is traversable, you might want to add a doc comment cast* to improve IDE auto-completion and static analysis: /** @var array $collection / $collection = json_decode($data, true); foreach ($collection as $item) { /* .. / } Mark the issue as a false-positive*: Just hover the remove button, in the top-right corner of this issue for more options. Loading history...
363			$actionParams = ['_index' => $index, '_type' => $type];
364
365			if (isset($doc['_id'])) {
366			$actionParams['_id'] = $doc['_id'];
367			unset($doc['_id']);
368			}
369
370			$bulkData .= json_encode([$action => $actionParams]) . "\n";
371			$bulkData .= json_encode($doc) . "\n";
372			}
373			break;
374			case 'update':
375			case 'delete':
376			foreach ($data as $doc) {
			0 ignored issues – show Bug introduced 2017-09-10 23:30 UTC by Report Bug Copy Issue Report The expression `$data` of type `null\|array` is not guaranteed to be traversable. How about adding an additional type check? There are different options of fixing this problem. If you want to be on the safe side, you can add an additional type-check: $collection = json_decode($data, true); if ( ! is_array($collection)) { throw new \RuntimeException('$collection must be an array.'); } foreach ($collection as $item) { /** ... / } If you are sure that the expression is traversable, you might want to add a doc comment cast* to improve IDE auto-completion and static analysis: /** @var array $collection / $collection = json_decode($data, true); foreach ($collection as $item) { /* .. / } Mark the issue as a false-positive*: Just hover the remove button, in the top-right corner of this issue for more options. Loading history...
377			if (!isset($doc['_id'])) {
378			throw new InvalidParamsException(
379			"_id field must be provided for each item in \$data param for $action action"
380			);
381			}
382
383			$actionParams = ['_index' => $index, '_type' => $type, '_id' => $doc['_id']];
384			unset($doc['_id']);
385
386			if ($action == 'update') {
387			if (!array_key_exists('doc', $doc)) {
388			$doc = ['doc' => $doc];
389			}
390
391			$bulkData .= json_encode([$action => $actionParams]) . "\n";
392			$bulkData .= json_encode($doc) . "\n";
393			}
394			}
395
396			break;
397			default:
398			$bulkActions = ['index', 'create', 'delete', 'update'];
399
400			throw new InvalidParamsException(
401			"Invalid 'action' param provided. Must be one of those: " . implode('\|', $bulkActions)
402			);
403			}
404
405			$response = $this->curlRequest->request("/_bulk", $bulkData, 'POST', [
406			'Content-Type: application/x-ndjson; charset=utf-8'
407			]);
408
409			if (isset($response['status']) && $response['status'] == 200) {
410			return $response['content'];
411			} elseif (isset($response['content'])) {
412			return $response['content'];
413			}
414
415			return false;
416			}
417
418			private function unsetEmpties(array $haystack) {
419			$selfFn = __FUNCTION__;
420
421			foreach ($haystack as $key => $value) {
422			if (is_array($value)) {
423			$haystack[$key] = $this->$selfFn($haystack[$key]);
424			}
425
426			if (empty($haystack[$key])) {
427			unset($haystack[$key]);
428			}
429			}
430
431			return $haystack;
432			}
433
434			/**
435			* @param string $index
436			* @return bool
437			*/
438			public function indexExists($index) {
439			$response = $this->curlRequest->request($index, [], 'HEAD');
440
441			return $response['status'] === 200;
442			}
443
444			/**
445			* @param string $index
446			* @param string $type
447			* @return bool
448			*/
449			public function typeExists($index, $type) {
450			$response = $this->curlRequest->request("$index/$type", [], 'HEAD');
451
452			return $response['status'] === 200;
453			}
454
455			private function throwExceptionFromResponse($response, $appendPrefix = '') {
456			if (isset($response['content']['error']['reason'])) {
457			if (!empty($appendPrefix)) {
458			$appendPrefix .= ': ';
459			}
460
461			throw new ElasticOperationException($appendPrefix . $response['content']['error']['reason']);
462			}
463			}
464
465			public function hasConnection() {
466			$response = $this->curlRequest->request('', [], 'HEAD');
467
468			return $response['status'] == 200;
469			}
470
471			private function setErrorFromElasticReturn($return) {
472			if (isset($return['error']['root_cause'][0]['reason'])) {
473			$this->setError($return['error']['root_cause'][0]['reason']);
474			} else if (isset($return['error']['reason'])) {
475			$this->setError($return['error']['reason']);
476			}
477			}
478
479			public function getElasticsearchVersion() {
480			if (is_null($this->esVersion)) {
481			$response = $this->curlRequest->request('', [], 'GET');
482
483			if (isset($response['content']['version']['number'])) {
484			$this->esVersion = floatval($response['content']['version']['number']);
485			} else {
486			throw new ConnectionException('Unable to fetch elasticsearch version. ');
487			}
488			}
489
490			return $this->esVersion;
491			}
492			}
493

andsalves / doctrine-elastic

GitHub Access Token became invalid

Push — master ( a53f28...120c3a )

ElasticConnection::createType() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like