Can the following code, which I copied from https://pypi.org/project/cloud-sql-python-connector/, be tailored to use Tortoise ORM?
import asyncio
import asyncpg
import sqlalchemy
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
from google.cloud.sql.connector import Connector
async def init_connection_pool(connector: Connector) -> AsyncEngine:
# initialize Connector object for connections to Cloud SQL
async def getconn() -> asyncpg.Connection:
conn: asyncpg.Connection = await connector.connect_async(
"project:region:instance", # Cloud SQL instance connection name
"asyncpg",
user="my-user",
password="my-password",
db="my-db-name"
# ... additional database driver args
)
return conn
# The Cloud SQL Python Connector can be used along with SQLAlchemy using the
# 'async_creator' argument to 'create_async_engine'
pool = create_async_engine(
"postgresql+asyncpg://",
async_creator=getconn,
)
return pool
async def main():
# initialize Connector object for connections to Cloud SQL
loop = asyncio.get_running_loop()
async with Connector(loop=loop) as connector:
# initialize connection pool
pool = await init_connection_pool(connector)
# example query
async with pool.connect() as conn:
await conn.execute(sqlalchemy.text("SELECT NOW()"))
# dispose of connection pool
await pool.dispose()
The above code currently uses SQLAlchemy and I'd much rather use Tortoise ORM.