-
Notifications
You must be signed in to change notification settings - Fork 4
/
Copy pathcypher_text2sql_view.py
42 lines (32 loc) · 1.48 KB
/
cypher_text2sql_view.py
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
41
42
# pylint: disable=missing-return-doc, missing-function-docstring, missing-class-docstring, missing-return-type-doc
from typing import List
import sqlalchemy
from sqlalchemy import text
from dbally.views.freeform.text2sql import BaseText2SQLView, ColumnConfig, TableConfig
class SampleText2SQLViewCyphers(BaseText2SQLView):
def get_tables(self) -> List[TableConfig]:
return [
TableConfig(
name="security_specialists",
columns=[
ColumnConfig("id", "SERIAL PRIMARY KEY"),
ColumnConfig("name", "VARCHAR(255)"),
ColumnConfig("cypher", "VARCHAR(255)"),
],
description="Knowledge base",
)
]
def create_freeform_memory_engine() -> sqlalchemy.Engine:
freeform_engine = sqlalchemy.create_engine("sqlite:///:memory:")
statements = [
"CREATE TABLE security_specialists (id INTEGER PRIMARY KEY, name TEXT, cypher TEXT)",
"INSERT INTO security_specialists (name, cypher) VALUES ('Alice', 'HAMAC')",
"INSERT INTO security_specialists (name, cypher) VALUES ('Bob', 'AES')",
"INSERT INTO security_specialists (name, cypher) VALUES ('Charlie', 'RSA')",
"INSERT INTO security_specialists (name, cypher) VALUES ('David', 'SHA2')",
]
with freeform_engine.connect() as conn:
for statement in statements:
conn.execute(text(statement))
conn.commit()
return freeform_engine