provision.py 文件源码

def _pg_create_db(cfg, eng, ident):
    with eng.connect().execution_options(
            isolation_level="AUTOCOMMIT") as conn:
        try:
            _pg_drop_db(cfg, conn, ident)
        except Exception:
            pass
        currentdb = conn.scalar("select current_database()")
        for attempt in range(3):
            try:
                conn.execute(
                    "CREATE DATABASE %s TEMPLATE %s" % (ident, currentdb))
            except exc.OperationalError as err:
                if attempt != 2 and "accessed by other users" in str(err):
                    time.sleep(.2)
                    continue
                else:
                    raise
            else:
                break