Код: Выделить всё
import aiomysql
import asyncio
loop = asyncio.get_event_loop()
async def fetch(cfg: dict):
records = []
if cfg["gmysql-enabled"]:
conn = None
try:
conn = await aiomysql.connect(
host=cfg['gmysql-host'], port=int(cfg['gmysql-port']),
user=cfg['gmysql-user'], password=cfg['gmysql-password'],
db=cfg['gmysql-dbname'], loop=loop
)
async with conn.cursor() as cur:
await cur.execute("""
SELECT domains.name, records.name, records.type, ttl, content, prio
FROM records INNER JOIN domains
WHERE records.domain_id=domains.id
""")
r = await cur.fetchall()
records = list(r)
except Exception as e:
finally:
if conn is not None:
await conn.ensure_closed()
return records
Код: Выделить всё
class AiomysqlCall extends DataFlow::Node {
AiomysqlCall() {
this = API::moduleImport("aiomysql").getMember("connect").getACall()
}
}
Код: Выделить всё
class AiomysqlCall extends DataFlow::Node {
AiomysqlCall() {
this = API::moduleImport("aiomysql").getMember("connect").getMember("cursor").getMember("execute").getACall()
}
}
Подробнее здесь: https://stackoverflow.com/questions/791 ... eql-python
Мобильная версия