db.py 文件源码

def __init__(self, db_url=None):
        self.db_url = db_url
        self.url = make_url(self.db_url)
        if self.backend == 'postgres':
            self.loop = asyncio.get_event_loop()
            self.async_engine = self.loop.run_until_complete(
                create_engine(
                    user=self.url.username,
                    password=self.url.password,
                    host=self.url.host,
                    port=self.url.port,
                    database=self.url.database))
            self.engine = sa.create_engine(self.db_url)
        else:
            self.engine = sa.create_engine(self.db_url)
        self.metadata = metadata
        self.metadata.create_all(bind=self.engine)

    # db helper methods