如何在sqlalchemy中编写自己的方言以适应HTTP API？ [英] How to write my own dialect in sqlalchemy to adapt HTTP API?
问题描述
我正在尝试向Superset（数据浏览平台）添加特殊的数据源。该数据库仅支持HTTP API，并以json格式返回数据；例如：
I'm trying to add a special data source to Superset (a data exploration platform). This database only supports HTTP API, and returns data in json format; for example:
> http://localhost/api/sql/query?q="select * from table"
< [{"id": 1, "value":10}, {"id": 2, "value": 30} ...]
因此，我必须在python SQLAlchemy for Superset中编写自己的适配器。我已经阅读了文档和部分源代码，但仍然需要一个很好的示例来进行跟踪。
Therefore, I have to write my own adapter in python SQLAlchemy for Superset. I have read the document and part of source code, but still need good example to follow.
推荐答案
（解决方案已编辑成问题由OP）
(Solution edited into the question by the OP)
我已经解决了这个问题。这就是我的工作。
I have solved this problem. Here is what I do.
-
转到
。/site-packages / sqlalchemy / dialects
将任何具体的方言复制到新的方言（例如，名为zeta）作为起点。更好的方法是使用
Copy any concrete dialects to the new one (eg: named zeta) as the start point. A better way is to use
from sqlalchemy.engine.default import DefaultDialect
class ZetaDialect(DefaultDialect):
...
将zeta添加到 __ all __ 的c $ c>部分。/site-packages/sqlalchemy/dialects/ __ init __。py
Add zeta into __all__ section of ./site-packages/sqlalchemy/dialects/__init__.py
创建一个测试程序：
from sqlalchemy import create_engine
engine = create_engine('zeta://XXX')
result = engine.execute("select * from table_name")
for row in result:
print(row)
-
运行它并获取错误。使用pdb查找原因。在大多数情况下，原因是NotImplement一些接口。
Run it and get error. Use pdb to find the reason. In most cases, the reason is NotImplement some interfaces. Solve it one by one.
当测试程序给出正确答案时，差不多已经完成了90％。为了完整起见，我们还应该实现检查员使用的几个接口：
When test program gives correct answer, it has almost been done 90%. For completeness, we should also implement several interface used by inspectors:
class ZetaDialect(DefaultDialect):
# default_paramstyle = 'qmark'
name = 'zeta'
def __init__(self, **kwargs):
DefaultDialect.__init__(self, **kwargs)
@classmethod
def dbapi(cls):
return zeta_dbapi
@reflection.cache
def get_table_names(self, connection, schema=None, **kw):
return [u'table_1', u'table_2', ...]
@reflection.cache
def get_pk_constraint(self, connection, table_name, schema=None, **kw):
return []
@reflection.cache
def get_foreign_keys(self, connection, table_name, schema=None, **kw):
return []
@reflection.cache
def get_unique_constraints(self, connection, table_name,
schema=None, **kw):
return []
@reflection.cache
def get_indexes(self, connection, table_name, schema=None, **kw):
return []
@reflection.cache
def get_schema_names(self, connection, **kw):
return []
@reflection.cache
def get_columns(self, connection, table_name, schema=None, **kw):
# just an example of the column structure
result = connection.execute('select * from %s limit 1' % table_name)
return [{'default': None, 'autoincrement': False, 'type': TEXT, 'name': colname, 'nullable': False} for colname, coltype in result.cursor.description]
这篇关于如何在sqlalchemy中编写自己的方言以适应HTTP API？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
