-
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
32 changed files
with
969 additions
and
179 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,20 @@ | ||
| <?php | ||
|
|
||
| declare(strict_types=1); | ||
|
|
||
| namespace SimPod\ClickHouseClient\Client; | ||
|
|
||
| use GuzzleHttp\Promise\PromiseInterface; | ||
| use SimPod\ClickHouseClient\Format\Format; | ||
| use SimPod\ClickHouseClient\Output\Output; | ||
|
|
||
| interface ClickHouseAsyncClient | ||
| { | ||
| /** | ||
| * @see Output hack for IDe to preserve `use` | ||
| * | ||
| * @phpstan-template O of Output | ||
| * @phpstan-param Format<O> $outputFormat | ||
| */ | ||
| public function select(string $sql, Format $outputFormat) : PromiseInterface; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| <?php | ||
|
|
||
| declare(strict_types=1); | ||
|
|
||
| namespace SimPod\ClickHouseClient\Client; | ||
|
|
||
| use Ds\Set; | ||
| use Ds\Vector; | ||
| use SimPod\ClickHouseClient\Format\Format; | ||
| use SimPod\ClickHouseClient\Output\Output; | ||
|
|
||
| interface ClickHouseClient | ||
| { | ||
| public function executeQuery(string $sql) : void; | ||
|
|
||
| /** | ||
| * @param array<string, int|string> $requestParameters | ||
| * | ||
| * @phpstan-template O of Output | ||
| * @phpstan-param Format<O> $outputFormat | ||
| * @phpstan-return O | ||
| */ | ||
| public function select(string $sql, Format $outputFormat, array $requestParameters = []) : Output; | ||
|
|
||
| /** | ||
| * @param Vector<mixed> $values | ||
| * @param Set<string>|null $columns | ||
| */ | ||
| public function insert(string $table, Vector $values, ?Set $columns = null) : void; | ||
|
|
||
| /** | ||
| * @phpstan-template O of Output | ||
| * @phpstan-param Format<O> $inputFormat | ||
| */ | ||
| public function insertWithFormat(string $table, Format $inputFormat, string $data) : void; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,93 @@ | ||
| <?php | ||
|
|
||
| declare(strict_types=1); | ||
|
|
||
| namespace SimPod\ClickHouseClient\Client; | ||
|
|
||
| use GuzzleHttp\Promise\PromiseInterface; | ||
| use Http\Client\HttpAsyncClient; | ||
| use Psr\Http\Message\ResponseInterface; | ||
| use SimPod\ClickHouseClient\Exception\ServerError; | ||
| use SimPod\ClickHouseClient\Format\Format; | ||
| use SimPod\ClickHouseClient\Http\RequestFactory; | ||
| use SimPod\ClickHouseClient\Http\RequestOptions; | ||
| use SimPod\ClickHouseClient\Output\Output; | ||
| use function GuzzleHttp\Promise\promise_for; | ||
|
|
||
| class PsrClickHouseAsyncClient implements ClickHouseAsyncClient | ||
| { | ||
| /** @var HttpAsyncClient */ | ||
| private $asyncClient; | ||
|
|
||
| /** @var RequestFactory */ | ||
| private $requestFactory; | ||
|
|
||
| /** @var string */ | ||
| private $endpoint; | ||
|
|
||
| /** @var array<string, int|string> */ | ||
| private $defaultParameters; | ||
|
|
||
| /** | ||
| * @param array<string, int|string> $defaultParameters | ||
| */ | ||
| public function __construct( | ||
| HttpAsyncClient $asyncClient, | ||
| RequestFactory $requestFactory, | ||
| string $endpoint, | ||
| array $defaultParameters = [] | ||
| ) { | ||
| $this->asyncClient = $asyncClient; | ||
| $this->requestFactory = $requestFactory; | ||
| $this->endpoint = $endpoint; | ||
| $this->defaultParameters = $defaultParameters; | ||
| } | ||
|
|
||
| public function select(string $sql, Format $outputFormat) : PromiseInterface | ||
| { | ||
| $formatClause = $outputFormat::toSql(); | ||
|
|
||
| return $this->executeRequest( | ||
| <<<CLICKHOUSE | ||
| $sql | ||
| $formatClause | ||
| CLICKHOUSE, | ||
| [], | ||
| static function (ResponseInterface $response) use ($outputFormat) : Output { | ||
| return $outputFormat::output((string) $response->getBody()); | ||
| } | ||
| ); | ||
| } | ||
|
|
||
| /** | ||
| * @param array<string, int|string> $requestParameters | ||
| * @param callable(ResponseInterface):mixed|null $processResponse | ||
| */ | ||
| private function executeRequest(string $sql, array $requestParameters = [], ?callable $processResponse = null) : PromiseInterface | ||
| { | ||
| $request = $this->requestFactory->prepareRequest( | ||
| $this->endpoint, | ||
| new RequestOptions( | ||
| $sql, | ||
| $this->defaultParameters, | ||
| $requestParameters | ||
| ) | ||
| ); | ||
|
|
||
| $promise = promise_for($this->asyncClient->sendAsyncRequest($request)); | ||
|
|
||
| return $promise->then( | ||
| static function (ResponseInterface $response) use ($processResponse) { | ||
| if ($response->getStatusCode() !== 200) { | ||
| throw ServerError::fromResponse($response); | ||
| } | ||
|
|
||
| if ($processResponse === null) { | ||
| return $response; | ||
| } | ||
|
|
||
| return $processResponse($response); | ||
| } | ||
| ); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,157 @@ | ||
| <?php | ||
|
|
||
| declare(strict_types=1); | ||
|
|
||
| namespace SimPod\ClickHouseClient\Client; | ||
|
|
||
| use Ds\Set; | ||
| use Ds\Vector; | ||
| use Psr\Http\Client\ClientInterface; | ||
| use Psr\Http\Message\ResponseInterface; | ||
| use SimPod\ClickHouseClient\Exception\CannotInsert; | ||
| use SimPod\ClickHouseClient\Exception\ServerError; | ||
| use SimPod\ClickHouseClient\Format\Format; | ||
| use SimPod\ClickHouseClient\Http\RequestFactory; | ||
| use SimPod\ClickHouseClient\Http\RequestOptions; | ||
| use SimPod\ClickHouseClient\Output\Output; | ||
| use function array_keys; | ||
| use function implode; | ||
| use function Safe\sprintf; | ||
|
|
||
| class PsrClickHouseClient implements ClickHouseClient | ||
| { | ||
| /** @var ClientInterface */ | ||
| private $client; | ||
|
|
||
| /** @var RequestFactory */ | ||
| private $requestFactory; | ||
|
|
||
| /** @var string */ | ||
| private $endpoint; | ||
|
|
||
| /** @var array<string, int|string> */ | ||
| private $defaultParameters; | ||
|
|
||
| /** | ||
| * @param array<string, int|string> $defaultParameters | ||
| */ | ||
| public function __construct( | ||
| ClientInterface $client, | ||
| RequestFactory $requestFactory, | ||
| string $endpoint, | ||
| array $defaultParameters = [] | ||
| ) { | ||
| $this->client = $client; | ||
| $this->requestFactory = $requestFactory; | ||
| $this->endpoint = $endpoint; | ||
| $this->defaultParameters = $defaultParameters; | ||
| } | ||
|
|
||
| public function executeQuery(string $sql) : void | ||
| { | ||
| $response = $this->executeRequest($sql); | ||
|
|
||
| $contents = (string) $response->getBody(); | ||
|
|
||
| return; | ||
| } | ||
|
|
||
| public function select(string $sql, Format $outputFormat, array $requestParameters = []) : Output | ||
| { | ||
| $formatClause = $outputFormat::toSql(); | ||
|
|
||
| $response = $this->executeRequest( | ||
| <<<CLICKHOUSE | ||
| $sql | ||
| $formatClause | ||
| CLICKHOUSE | ||
| ); | ||
|
|
||
| return $outputFormat::output((string) $response->getBody()); | ||
| } | ||
|
|
||
| public function insert(string $table, Vector $values, ?Set $columns = null) : void | ||
| { | ||
| if ($values->isEmpty()) { | ||
| throw CannotInsert::noValues(); | ||
| } | ||
|
|
||
| if ($columns === null) { | ||
| /** @var array<mixed> $row */ | ||
| $row = $values->first(); | ||
| $columnNames = array_keys($row); | ||
| } else { | ||
| $columnNames = $columns->toArray(); | ||
| } | ||
|
|
||
| $columnsSql = implode(',', $columnNames); | ||
|
|
||
| $valuesSql = implode( | ||
| ',', | ||
| $values->map( | ||
| static function (array $map) : string { | ||
| return sprintf( | ||
| '(%s)', | ||
| implode(',', $map) | ||
| ); | ||
| } | ||
| )->toArray() | ||
| ); | ||
|
|
||
| $response = $this->executeRequest( | ||
| <<<CLICKHOUSE | ||
| INSERT INTO $table | ||
| ($columnsSql) | ||
| VALUES $valuesSql | ||
| CLICKHOUSE | ||
| ); | ||
| } | ||
|
|
||
| public function insertWithFormat(string $table, Format $inputFormat, string $data) : void | ||
| { | ||
| $formatSql = $inputFormat::toSql(); | ||
|
|
||
| $this->executeRequest( | ||
| <<<CLICKHOUSE | ||
| INSERT INTO $table $formatSql $data | ||
| CLICKHOUSE | ||
| ); | ||
| } | ||
|
|
||
| /** | ||
| * @param array<string, int|string> $requestParameters | ||
| */ | ||
| private function executeRequest(string $sql, array $requestParameters = []) : ResponseInterface | ||
| { | ||
| $request = $this->requestFactory->prepareRequest( | ||
| $this->endpoint, | ||
| new RequestOptions( | ||
| $sql, | ||
| $this->defaultParameters, | ||
| $requestParameters | ||
| ) | ||
| ); | ||
| $response = $this->client->sendRequest($request); | ||
| if ($response->getStatusCode() !== 200) { | ||
| throw ServerError::fromResponse($response); | ||
| } | ||
|
|
||
| return $response; | ||
| } | ||
|
|
||
| /** | ||
| * @return array<string, int|string> | ||
| */ | ||
| public function getDefaultParameters() : array | ||
| { | ||
| return $this->defaultParameters; | ||
| } | ||
|
|
||
| /** | ||
| * @param array<string, int|string> $defaultParameters | ||
| */ | ||
| public function setDefaultParameters(array $defaultParameters) : void | ||
| { | ||
| $this->defaultParameters = $defaultParameters; | ||
| } | ||
| } |
Oops, something went wrong.