forked from kelsin/mysql-mimic
-
Notifications
You must be signed in to change notification settings - Fork 1
/
simple.py
43 lines (32 loc) · 765 Bytes
/
simple.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
43
import logging
import asyncio
from sqlglot.executor import execute
from mysql_mimic import MysqlServer, Session
SCHEMA = {
"test": {
"x": {
"a": "INT",
}
}
}
TABLES = {
"test": {
"x": [
{"a": 1},
{"a": 2},
{"a": 3},
]
}
}
class MySession(Session):
async def query(self, expression, sql, attrs):
result = execute(expression, schema=SCHEMA, tables=TABLES)
return result.rows, result.columns
async def schema(self):
return SCHEMA
async def main():
logging.basicConfig(level=logging.DEBUG)
server = MysqlServer(session_factory=MySession)
await server.serve_forever()
if __name__ == "__main__":
asyncio.run(main())