blob: b7eee4ad7db99c447732e3f3ebf2e8c108fe93a8 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
|
from sqlalchemy.sql.ddl import DDL
from sqlalchemy.exc import ArgumentError
from clickhouse_connect.driver.query import quote_identifier
# pylint: disable=too-many-ancestors,abstract-method
class CreateDatabase(DDL):
"""
SqlAlchemy DDL statement that is essentially an alternative to the built in CreateSchema DDL class
"""
# pylint: disable-msg=too-many-arguments
def __init__(self, name: str, engine: str = None, zoo_path: str = None, shard_name: str = '{shard}',
replica_name: str = '{replica}'):
"""
:param name: Database name
:param engine: Database ClickHouse engine type
:param zoo_path: ClickHouse zookeeper path for Replicated database engine
:param shard_name: Clickhouse shard name for Replicated database engine
:param replica_name: Replica name for Replicated database engine
"""
if engine and engine not in ('Ordinary', 'Atomic', 'Lazy', 'Replicated'):
raise ArgumentError(f'Unrecognized engine type {engine}')
stmt = f'CREATE DATABASE {quote_identifier(name)}'
if engine:
stmt += f' Engine {engine}'
if engine == 'Replicated':
if not zoo_path:
raise ArgumentError('zoo_path is required for Replicated Database Engine')
stmt += f" ('{zoo_path}', '{shard_name}', '{replica_name}'"
super().__init__(stmt)
# pylint: disable=too-many-ancestors,abstract-method
class DropDatabase(DDL):
"""
Alternative DDL statement for built in SqlAlchemy DropSchema DDL class
"""
def __init__(self, name: str):
super().__init__(f'DROP DATABASE {quote_identifier(name)}')
|
