sqlalchemy async запрос для mysql
как можно сделать URL на создание асинхронного движка? что нужно писать для драйвера в функции DATABASE_URL_async?
Код движка
engine = create_async_engine(
url=settings.DATABASE_URL,
echo=False
)
with engine.connect() as con:
res = con.execute(text('SELECT VERSION()'))
print(f'{res.all()=}')
Код создания URL
from dotenv import dotenv_values
class Settings():
DB_HOST: str
DB_PORT: int
DB_USER: str
DB_PASS: str
DB_NAME: str
def __init__(self):
self.DB_HOST = data.get('DB_HOST')
self.DB_PORT = data.get('DB_PORT')
self.DB_USER = data.get('DB_USER')
self.DB_PASS = data.get('DB_PASS')
self.DB_NAME = data.get('DB_NAME')
@property
def DATABASE_URL_asynс(self):
return f"mysql://{self.DB_USER}:{self.DB_PASS}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
@property
def DATABASE_URL(self):
return f"mysql+pymysql://{self.DB_USER}:{self.DB_PASS}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
data = dotenv_values()
settings = Settings()