Python sqlalchemy_utils 模块，create_database() 实例源码

def _create_new_database(cls, url):
        """Used by testing to create a new database."""
        purl = sqlalchemy_url.make_url(
            cls.dress_url(
                url))
        purl.database = purl.database + str(uuid.uuid4()).replace('-', '')
        new_url = str(purl)
        sqlalchemy_utils.create_database(new_url)
        return new_url

def init_db():
    if sqlalchemy_utils.database_exists(engine.url):
        sqlalchemy_utils.drop_database(engine.url)
    sqlalchemy_utils.create_database(engine.url)
    print("DB ??? ??")

def init_db():
    if sqlalchemy_utils.database_exists(engine.url):
        sqlalchemy_utils.drop_database(engine.url)
    sqlalchemy_utils.create_database(engine.url)
    print("DB ??? ??")

def start_fixture(self):
        """Set up config."""

        global LOAD_APP_KWARGS

        self.conf = None

        # Determine the database connection.
        db_url = os.environ.get('PIFPAF_URL', "sqlite://").replace(
            "mysql://", "mysql+pymysql://")
        if not db_url:
            raise case.SkipTest('No database connection configured')

        conf = self.conf = service.prepare_service([], [])
        opts.set_defaults(self.conf)

        content = ('{"default": ""}')
        if six.PY3:
            content = content.encode('utf-8')
        self.tempfile = fileutils.write_to_tempfile(content=content,
                                                    prefix='policy',
                                                    suffix='.json')

        conf.set_override("policy_file", self.tempfile,
                          group='oslo_policy')
        conf.set_override(
            'api_paste_config',
            os.path.abspath('etc/panko/api_paste.ini')
        )

        parsed_url = list(urlparse.urlparse(db_url))
        parsed_url[2] += '-%s' % uuidutils.generate_uuid(dashed=False)
        db_url = urlparse.urlunparse(parsed_url)

        conf.set_override('connection', db_url, group='database')

        if (parsed_url[0].startswith("mysql")
           or parsed_url[0].startswith("postgresql")):
            sqlalchemy_utils.create_database(conf.database.connection)

        self.conn = storage.get_connection_from_config(self.conf)
        self.conn.upgrade()

        LOAD_APP_KWARGS = {
            'conf': conf, 'appname': 'panko+noauth',
        }