• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Python functions.drop_database函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中sqlalchemy_utils.functions.drop_database函数的典型用法代码示例。如果您正苦于以下问题:Python drop_database函数的具体用法?Python drop_database怎么用?Python drop_database使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



在下文中一共展示了drop_database函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: populate

def populate(bulk_data, test_data, posts, topics, force, initdb):
    """Creates the necessary tables and groups for FlaskBB."""
    if force:
        click.secho("[+] Recreating database...", fg="cyan")
        drop_database(db.engine.url)

        # do not initialize the db if -i is passed
        if not initdb:
            upgrade_database()

    if initdb:
        click.secho("[+] Initializing database...", fg="cyan")
        upgrade_database()

    if test_data:
        click.secho("[+] Adding some test data...", fg="cyan")
        create_test_data()

    if bulk_data:
        timer = time.time()
        topic_count, post_count = insert_bulk_data(int(topics), int(posts))
        elapsed = time.time() - timer
        click.secho("[+] It took {} seconds to create {} topics and {} posts"
                    .format(elapsed, topic_count, post_count), fg="cyan")

    # this just makes the most sense for the command name; use -i to
    # init the db as well
    if not test_data:
        click.secho("[+] Populating the database with some defaults...",
                    fg="cyan")
        create_default_groups()
        create_default_settings()
开发者ID:djsilcock,项目名称:flaskbb,代码行数:32,代码来源:main.py


示例2: install

def install(welcome, force, username, email, password, group):
    """Installs flaskbb. If no arguments are used, an interactive setup
    will be run.
    """
    click.secho("[+] Installing FlaskBB...", fg="cyan")
    if database_exists(db.engine.url):
        if force or click.confirm(click.style(
            "Existing database found. Do you want to delete the old one and "
            "create a new one?", fg="magenta")
        ):
            drop_database(db.engine.url)
        else:
            sys.exit(0)
    create_database(db.engine.url)
    upgrade_database()

    click.secho("[+] Creating default settings...", fg="cyan")
    create_default_groups()
    create_default_settings()

    click.secho("[+] Creating admin user...", fg="cyan")
    prompt_save_user(username, email, password, group)

    if welcome:
        click.secho("[+] Creating welcome forum...", fg="cyan")
        create_welcome_forum()

    click.secho("[+] Compiling translations...", fg="cyan")
    compile_translations()

    click.secho("[+] FlaskBB has been successfully installed!",
                fg="green", bold=True)
开发者ID:djsilcock,项目名称:flaskbb,代码行数:32,代码来源:main.py


示例3: test_alembic_and_db_create_match

def test_alembic_and_db_create_match(clean_app):
    """Check that alembic recipes and alembic models are in sync."""
    with clean_app.app_context():
        ext = clean_app.extensions['invenio-db']
        if db.engine.name == 'sqlite':
            raise pytest.skip('Upgrades are not supported on SQLite.')

        # Make sure that alembic upgrades can be run now
        ext.alembic.upgrade()
        constraints = get_all_constraints(clean_app)
        alembic_constraint_names = [x[0] for x in constraints]

        # Recreate the database with alembic only (without migrating)
        db.session.remove()
        drop_database(db.engine.url)
        db.engine.dispose()
        create_database(db.engine.url)
        db.create_all()

        # Check that the resulting state is in sync with alembic metaData
        assert not ext.alembic.compare_metadata()

        # Check that the constraints are the same. This is needed because
        # alembic.compare_metadata does not check all constraints.
        constraints = get_all_constraints(clean_app)
        db_create_constraint_names = [x[0] for x in constraints]
        assert set(alembic_constraint_names) == set(db_create_constraint_names)
开发者ID:EUDAT-B2SHARE,项目名称:b2share,代码行数:27,代码来源:test_upgrade.py


示例4: clean_app

def clean_app(request, base_app):
    """Application with database and elasticsearch cleaned."""
    with base_app.app_context():
        try:
            db.session.remove()
            drop_database(db.engine.url)
        except ProgrammingError:
            pass
        create_database(db.engine.url)
        # reset elasticsearch
        for deleted in current_search.delete(ignore=[404]):
            pass
        # reset queues
        current_queues.delete()
        current_queues.declare()

    yield base_app

    def finalize():
        with base_app.app_context():
            db.session.remove()
            drop_database(db.engine.url)
            # Dispose the engine in order to close all connections. This is
            # needed for sqlite in memory databases.
            db.engine.dispose()
            current_queues.delete()
    request.addfinalizer(finalize)

    return base_app
开发者ID:EUDAT-B2SHARE,项目名称:b2share,代码行数:29,代码来源:conftest.py


示例5: tearDownClass

 def tearDownClass(cls):
     cls.transaction.rollback()
     cls.connection.close()
     cls.engine.dispose()
     if cls.create:
         drop_database(cls.engine.url)
     super(StoreTestCase, cls).tearDownClass()
开发者ID:deliveryhero,项目名称:lymph-sqlalchemy,代码行数:7,代码来源:testing.py


示例6: base_app

def base_app():
    """Flask application fixture."""
    instance_path = tempfile.mkdtemp()
    base_app = Flask(__name__, instance_path=instance_path)

    base_app.config.update(
        ACCOUNTS_USE_CELERY=False,
        LOGIN_DISABLED=False,
        SECRET_KEY='testing_key',
        SQLALCHEMY_DATABASE_URI=os.getenv('SQLALCHEMY_DATABASE_URI',
                                          'sqlite://'),
        TEST_USER_EMAIL='[email protected]',
        TEST_USER_PASSWORD='test_password',
        TESTING=True,
        WTF_CSRF_ENABLED=False,
    )
    Babel(base_app)
    Mail(base_app)
    Menu(base_app)
    InvenioDB(base_app)
    InvenioAccounts(base_app)
    base_app.register_blueprint(accounts_blueprint)

    with base_app.app_context():
        if str(db.engine.url) != "sqlite://" and \
                not database_exists(str(db.engine.url)):
            create_database(str(db.engine.url))
        db.create_all()

    yield base_app

    with base_app.app_context():
        drop_database(str(db.engine.url))
    shutil.rmtree(instance_path)
开发者ID:inveniosoftware,项目名称:invenio-userprofiles,代码行数:34,代码来源:conftest.py


示例7: teardown

 def teardown():
     with app.app_context():
         drop_database(str(db.engine.url))
         # Delete sessions in kvsession store
         if hasattr(app, 'kvsession_store'):
             for key in app.kvsession_store.iter_keys():
                 app.kvsession_store.delete(key)
     shutil.rmtree(instance_path)
开发者ID:eamonnmag,项目名称:invenio-accounts,代码行数:8,代码来源:conftest.py


示例8: engine

def engine(dsn):
    if database_exists(dsn):
        drop_database(dsn)
    create_database(dsn)
    engine = create_engine(dsn)
    engine.execute('CREATE EXTENSION postgis')
    GeonameBase.metadata.create_all(bind=engine)
    return engine
开发者ID:jmagnusson,项目名称:sqlalchemy-geonames,代码行数:8,代码来源:test_base.py


示例9: teardown

 def teardown():
     with app.app_context():
         drop_database(str(db.engine.url))
         # Delete sessions in kvsession store
         if hasattr(app, 'kvsession_store') and \
                 isinstance(app.kvsession_store, RedisStore):
             app.kvsession_store.redis.flushall()
     shutil.rmtree(app.instance_path)
开发者ID:lnielsen,项目名称:invenio-accounts,代码行数:8,代码来源:conftest.py


示例10: db

def db(app):
    """Database fixture."""
    if not database_exists(str(db_.engine.url)):
        create_database(str(db_.engine.url))
    db_.create_all()
    yield db_
    db_.session.remove()
    drop_database(str(db_.engine.url))
开发者ID:inveniosoftware,项目名称:invenio-records-files,代码行数:8,代码来源:conftest.py


示例11: finalize

 def finalize():
     with base_app.app_context():
         db.session.remove()
         drop_database(db.engine.url)
         # Dispose the engine in order to close all connections. This is
         # needed for sqlite in memory databases.
         db.engine.dispose()
         current_queues.delete()
开发者ID:EUDAT-B2SHARE,项目名称:b2share,代码行数:8,代码来源:conftest.py


示例12: test_db

def test_db():
    """Test database backend."""
    app = Flask(__name__)
    app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get(
        'SQLALCHEMY_DATABASE_URI', 'sqlite:///test.db'
    )
    FlaskCLI(app)
    InvenioDB(app)
    InvenioRecords(app)

    with app.app_context():
        create_database(db.engine.url)
        db.create_all()
        assert len(db.metadata.tables) == 3

    data = {'title': 'Test'}
    from invenio_records.models import RecordMetadata as RM

    # Create a record
    with app.app_context():
        assert RM.query.count() == 0

        record_uuid = Record.create(data).id
        db.session.commit()

        assert RM.query.count() == 1
        db.session.commit()

    # Retrieve created record
    with app.app_context():
        record = Record.get_record(record_uuid)
        assert record.dumps() == data
        with pytest.raises(NoResultFound):
            Record.get_record(uuid.uuid4())
        record['field'] = True
        record = record.patch([
            {'op': 'add', 'path': '/hello', 'value': ['world']}
        ])
        assert record['hello'] == ['world']
        record.commit()
        db.session.commit()

    with app.app_context():
        record2 = Record.get_record(record_uuid)
        assert record2.model.version_id == 2
        assert record2['field']
        assert record2['hello'] == ['world']
        db.session.commit()

    # Cannot commit record without model (i.e. Record.create_record)
    with app.app_context():
        record3 = Record({'title': 'Not possible'})
        with pytest.raises(RecordNotCommitableError):
            record3.commit()

    with app.app_context():
        db.drop_all()
        drop_database(db.engine.url)
开发者ID:JavierDelgadoFernandez,项目名称:invenio-records,代码行数:58,代码来源:test_invenio_records.py


示例13: app

def app():
    """Flask application fixture."""
    instance_path = tempfile.mkdtemp()
    app = Flask('testapp', instance_path=instance_path)
    app.config.update(
        CELERY_ALWAYS_EAGER=True,
        CELERY_CACHE_BACKEND='memory',
        CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
        CELERY_RESULT_BACKEND='cache',
        JSONSCHEMAS_HOST='inveniosoftware.org',
        TESTING=True,
        SECRET_KEY='CHANGE_ME',
        SQLALCHEMY_DATABASE_URI=os.environ.get('SQLALCHEMY_DATABASE_URI',
                                               'sqlite://'),
        SQLALCHEMY_TRACK_MODIFICATIONS=True,
        SERVER_NAME='app',
        OAISERVER_RECORD_INDEX='_all',
        # Disable set signals because the celery tasks cannot be run
        # synchronously
        OAISERVER_REGISTER_SET_SIGNALS=False,
        SEARCH_ELASTIC_KEYWORD_MAPPING={None: ['_all']},
    )
    if not hasattr(app, 'cli'):
        from flask_cli import FlaskCLI
        FlaskCLI(app)
    InvenioDB(app)
    FlaskCeleryExt(app)
    InvenioJSONSchemas(app)
    InvenioRecords(app)
    InvenioPIDStore(app)
    InvenioMARC21(app)
    client = Elasticsearch(hosts=[os.environ.get('ES_HOST', 'localhost')])
    search = InvenioSearch(app, client=client)
    search.register_mappings('records', 'data')
    InvenioIndexer(app)
    InvenioOAIServer(app)

    app.register_blueprint(blueprint)

    with app.app_context():
        if str(db.engine.url) != 'sqlite://' and \
           not database_exists(str(db.engine.url)):
                create_database(str(db.engine.url))
        db.create_all()
        list(search.create(ignore=[400]))
        sleep(5)

    with app.app_context():
        yield app

    with app.app_context():
        db.session.close()
        if str(db.engine.url) != 'sqlite://':
            drop_database(str(db.engine.url))
        list(search.delete(ignore=[404]))
    shutil.rmtree(instance_path)
开发者ID:inveniosoftware,项目名称:invenio-oaiserver,代码行数:56,代码来源:conftest.py


示例14: database

def database(app):
    """Ensure that the database schema is created."""
    if not database_exists(str(db_.engine.url)):
        create_database(str(db_.engine.url))
    db_.drop_all()
    db_.create_all()

    yield db_

    drop_database(str(db_.engine.url))
开发者ID:duncanwp,项目名称:zenodo,代码行数:10,代码来源:conftest.py


示例15: test_migrations_upgrade

def test_migrations_upgrade():
    with pytest.raises(OperationalError):
        User.query.all()

    # ensure that the database is created
    create_database(db.engine.url)

    alembic.upgrade()
    assert len(User.query.all()) == 0

    drop_database(db.engine.url)
开发者ID:eirnym,项目名称:flaskbb,代码行数:11,代码来源:test_populate.py


示例16: db

def db(app, request):
    """Setup database."""
    if not database_exists(str(db_.engine.url)):
        create_database(str(db_.engine.url))
    db_.create_all()

    yield db_

    drop_database(str(db_.engine.url))
    # Delete sessions in kvsession store
    if hasattr(app, 'kvsession_store') and \
            isinstance(app.kvsession_store, RedisStore):
        app.kvsession_store.redis.flushall()
开发者ID:hachreak,项目名称:zenodo-accessrequests,代码行数:13,代码来源:conftest.py


示例17: db

def db(app):
    """Database fixture."""
    if not database_exists(str(db_.engine.url)) and \
            app.config['SQLALCHEMY_DATABASE_URI'] != 'sqlite://':
        create_database(db_.engine.url)
    db_.create_all()

    yield db_

    db_.session.remove()
    db_.session.close()
    if str(db_.engine.url) != 'sqlite://':
        drop_database(str(db_.engine.url))
开发者ID:inveniosoftware,项目名称:invenio-communities,代码行数:13,代码来源:conftest.py


示例18: init

def init(user='root', password='', yes_i_know=False):
    """Initialize database and user."""
    from invenio.ext.sqlalchemy.utils import initialize_database_user
    from invenio.utils.text import wrap_text_in_a_box, wait_for_user

    from sqlalchemy_utils.functions import database_exists, create_database, \
        drop_database

    from sqlalchemy.engine.url import URL
    from sqlalchemy import create_engine

    # Step 0: confirm deletion
    wait_for_user(wrap_text_in_a_box(
        "WARNING: You are going to destroy your database tables! Run first"
        " `inveniomanage database drop`."
    ))

    # Step 1: create URI to connect admin user
    cfg = current_app.config
    SQLALCHEMY_DATABASE_URI = URL(
        cfg.get('CFG_DATABASE_TYPE', 'mysql'),
        username=user,
        password=password,
        host=cfg.get('CFG_DATABASE_HOST'),
        database=cfg.get('CFG_DATABASE_NAME'),
        port=cfg.get('CFG_DATABASE_PORT'),
    )

    # Step 2: drop the database if already exists
    if database_exists(SQLALCHEMY_DATABASE_URI):
        drop_database(SQLALCHEMY_DATABASE_URI)
        print('>>> Database has been dropped.')

    # Step 3: create the database
    create_database(SQLALCHEMY_DATABASE_URI, encoding='utf8')
    print('>>> Database has been created.')

    # Step 4: setup connection with special user
    engine = create_engine(SQLALCHEMY_DATABASE_URI)
    engine.connect()

    # Step 5: grant privileges for the user
    initialize_database_user(
        engine=engine,
        database_name=current_app.config['CFG_DATABASE_NAME'],
        database_user=current_app.config['CFG_DATABASE_USER'],
        database_pass=current_app.config['CFG_DATABASE_PASS'],
    )
    print('>>> Database user has been initialized.')
开发者ID:SCOAP3,项目名称:invenio,代码行数:49,代码来源:database.py


示例19: update_database

def update_database():
    """Insert all cards in database"""

    new_version = fetch.mtgjson_version()
    current_version = get_version()
    if current_version is not None  and  new_version == current_version:
        return

    # TODO: create the new database in a different file and move to the
    # current file when the prices are updated
    if database_exists(config.DATABASE['db.url']):
        drop_database(engine.url)

    db_connection = engine.connect()
    schema.create_database_schema()
    cards = fetch.all_cards()
    for name, card in cards.items():
        insert_card(name, card)

    from database import price_grabber
    price_grabber.make_all_cards_list()
    update_version(db_connection, new_version)
开发者ID:arturojosejr,项目名称:heirloom,代码行数:22,代码来源:helper.py


示例20: validate_database_schema

def validate_database_schema(app, alembic):
    """Check that the database schema is as expected after a full migration."""
    # Check that the resulting state is in sync with sqlalchemy's MetaData
    assert not alembic.compare_metadata()

    # list all existing constraints
    current_constraints = get_all_constraints(app)
    current_constraint_names = [x[1] for x in current_constraints]

    # Recreate the database with alembic only (without migrating)
    db.session.remove()
    drop_database(db.engine.url)
    db.engine.dispose()
    create_database(db.engine.url)

    alembic.upgrade()

    # Check that the constraints are the same. This is needed because
    # alembic.compare_metadata does not check all constraints.
    expected_constraints = get_all_constraints(app)
    expected_constraint_names = [x[1] for x in expected_constraints]
    assert set(current_constraint_names) == set(expected_constraint_names)
开发者ID:emanueldima,项目名称:b2share,代码行数:22,代码来源:helpers.py



注:本文中的sqlalchemy_utils.functions.drop_database函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python sqlalchemy_wrapper.SQLAlchemy类代码示例发布时间:2022-05-27
下一篇:
Python functions.database_exists函数代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap