Commit 9dee79c7 authored by Lukáš Lalinský's avatar Lukáš Lalinský


parent cb615849
SQLAlchemy dialect implementation for Phoenix.
import urlparse
import urllib
import phoenixdb
from sqlalchemy.engine.default import DefaultDialect
from sqlalchemy.sql.compiler import DDLCompiler
from sqlalchemy.exc import CompileError
class PhoenixDDLCompiler(DDLCompiler):
def visit_primary_key_constraint(self, constraint):
if is None:
raise CompileError("can't create primary key without a name")
return DDLCompiler.visit_primary_key_constraint(self, constraint)
class PhoenixDialect(DefaultDialect):
name = "phoenix"
driver = "phoenixdb"
ddl_compiler = PhoenixDDLCompiler
def dbapi(cls):
return phoenixdb
def create_connect_args(self, url):
phoenix_url = urlparse.urlunsplit(urlparse.SplitResult(
netloc='{}:{}'.format(, url.port or 8765),
return [phoenix_url], {'autocommit': True}
def do_rollback(self, dbapi_conection):
def do_commit(self, dbapi_conection):
def has_table(self, connection, table_name, schema=None):
if schema is None:
query = "SELECT 1 FROM system.catalog WHERE table_name = ? LIMIT 1"
params = [table_name.upper()]
query = "SELECT 1 FROM system.catalog WHERE table_name = ? AND schema_name = ? LIMIT 1"
params = [table_name.upper(), schema_name.upper()]
return connection.execute(query, params).first() is not None
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment