Всем привет у меня есть запрос
query = insert(balance)
query = query.on_conflict_do_nothing(
index_elements=["query_id", "warehouse", "product"]
)
я использую sqlalchemy compile
sql_str = (
query.compile(
dialect=postgresql.dialect(),
)
)
на выходе получаю вот такую строку
INSERT INTO balance (id, query_id, warehouse, product) VALUES (%(id)s, %(query_id)s, %(warehouse)s, %(product)s) ON CONFLICT (query_id, warehouse, product) DO NOTHING
как мне получить вот такую raw sql
INSERT INTO balance (id, query_id, warehouse, product) VALUES ($1, $2, $3, $4) ON CONFLICT (query_id, warehouse, product) DO NOTHING
full code
import asyncio
import logging
import sys
from contextlib import asynccontextmanager
from typing import AsyncGenerator
import async_timeout
import asyncpgsa
import stackprinter
from faker import Faker
from sqlalchemy.dialects import postgresql
logging.basicConfig(level=logging.DEBUG)
import sqlalchemy
from sqlalchemy import Boolean, Column, Table, MetaData, NUMERIC, func
from sqlalchemy.dialects.postgresql import UUID, TIMESTAMP, insert
from sqlalchemy.types import Integer
metadata = MetaData()
stock_balance: sqlalchemy.table = Table(
"stock_balance",
metadata,
Column("warehouse", UUID, primary_key=True),
Column("product", UUID, primary_key=True),
Column("balance", NUMERIC(15, 3), index=True),
Column("reserve", NUMERIC(15, 3), index=True),
)
balance: sqlalchemy.table = Table(
"balance",
metadata,
Column("id", Integer, primary_key=True),
Column("query_id", UUID(as_uuid=True), nullable=False, index=True),
Column("warehouse", UUID, index=True, nullable=False),
Column("product", UUID, index=True, nullable=False),
Column("balance", Boolean, index=True, server_default="f"),
Column("reserve", Boolean, index=True, server_default="f"),
Column("count", NUMERIC(15, 3), nullable=False),
Column("date_time", TIMESTAMP, server_default=func.now(tz="UTC"), index=True),
Column("updated", Boolean, server_default="f", index=True),
)
def data_traffic():
fake = Faker("ru_RU")
balance = fake.boolean()
return {
"id": fake.uuid4(),
"warehouse": fake.uuid4(),
"product": fake.uuid4(),
"balance": balance,
"count": fake.pyfloat(left_digits=15, right_digits=3, min_value=0),
"reserve": not balance,
}
@asynccontextmanager
async def connect_db() -> AsyncGenerator:
try:
with async_timeout.timeout(5):
conn = await asyncpgsa.create_pool(
f"postgresql://postgres:some_secret@"
f"localhost:10001/stockbalance_test",
# echo=True,
min_size=1,
max_size=1,
dialect=postgresql.dialect()
)
async with conn.acquire() as c:
yield c
await conn.close()
except Exception as exc:
logging.error(
"Server Errors: {}\n{}\n{}\n{}".format(
exc, sys.exc_info()[0], sys.exc_info()[1], stackprinter.format()
)
)
yield None
finally:
await conn.close()
async def update_balance(conn, data: list):
query = insert(balance)
query = query.on_conflict_do_nothing(
index_elements=["query_id", "warehouse", "product"]
)
sql_str = (
query.compile(
dialect=postgresql.dialect(),
)
)
await conn.executemany(
str(sql_str),
[
(
item["id"],
item["warehouse"],
item["product"],
item["balance"],
item["reserve"],
item["count"],
)
for item in data
],
)
async def main():
async with connect_db() as conn:
try:
await update_balance(conn, [data_traffic()])
except Exception as exc:
print(exc)
if __name__ == "__main__":
asyncio.run(main())
syntax error at or near "%"