python类create_all()的实例源码

def initdb():
    """
    init database, create all tables, create user plan and create admin.
    """
    print("init database...")
    try:
        from app.model.user import UserPlan, User, PlanUsage
        from app.model.info import WebSite, Article, SiteType, Tag, tags
        from app.model.subscribe import UserSub
        db.create_all()
    except Exception as e:
        print e

    try:
        init_user()
    except Exception:
        pass

    print "finish."

def setUp(self):
        db.drop_all()
        db.create_all()
        connection = db.engine.raw_connection()
        try:
            c = connection.cursor()

            for f in self.INIT_FILES:
                script = open(
                    os.path.join(os.path.dirname(__file__), self.SQL_DIR, f), 'rt', encoding='utf-8-sig').read()
                # print(script)
                res = c.execute(script)
                connection.commit()

        finally:
            pass
            connection.close()

def setUp(self):
        """
        Will be called before every test
        """

        db.create_all()

        # create test admin user
        admin = Employee(username="admin", password="admin2016", is_admin=True)

        # create test non-admin user
        employee = Employee(username="test_user", password="test2016")

        # save users to database
        db.session.add(admin)
        db.session.add(employee)
        db.session.commit()

def before_all(context):
    """Create the context bag."""
    # Change the configuration on the fly
    os.environ['ENV'] = 'testing'

    # Import this now because the environment variable
    # should be changed before this happens.

    context.app = create_app()

    # Create all the tables in memory.
    with context.app.app_context():
        db.create_all()

    # Create the test client.
    context.client = context.app.test_client()

def setUp(self):
        """
        Will be called before every test
        """
        db.session.commit()
        db.drop_all()
        db.create_all()

        # create test admin user
        admin = Employee(username="admin", password="admin2016", is_admin=True)

        # create test non-admin user
        employee = Employee(username="test_user", password="test2016")

        # save users to database
        db.session.add(admin)
        db.session.add(employee)
        db.session.commit()

def initdb(ctx):
    from app import app, db
    with app.app_context():
        db.create_all()

def resetdb(ctx):
    from app import app, db
    with app.app_context():
        db.drop_all()
        db.create_all()

def to_dict_v2(self):
        response = {
            "doi": self.doi,
            "doi_url": self.url,
            "is_oa": self.is_oa,
            "best_oa_location": self.best_oa_location_dict,
            "oa_locations": self.all_oa_location_dicts(),
            "data_standard": self.data_standard,
            "title": self.best_title,
            "year": self.year,
            "journal_is_oa": self.oa_is_open_journal,
            "journal_is_in_doaj": self.oa_is_doaj_journal,
            "journal_issns": self.display_issns,
            "journal_name": self.journal,
            "publisher": self.publisher,
            "published_date": self.issued,
            "updated": self.display_updated,
            "genre": self.genre,
            "z_authors": self.authors,
            # "crossref_api_modified": self.crossref_api_modified,

            # need this one for Unpaywall
            "x_reported_noncompliant_copies": self.reported_noncompliant_copies,

            # "x_crossref_api_raw": self.crossref_api_modified

        }

        if self.error:
            response["x_error"] = True

        return response





# db.create_all()
# commit_success = safe_commit(db)
# if not commit_success:
#     logger.info(u"COMMIT fail making objects")

def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
        Role.insert_roles()
        self.client = self.app.test_client(use_cookies=True)

def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
        Role.insert_roles()
        self.client = self.app.test_client()

def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()

def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
        Role.insert_roles()

def setUp(self):
        self.ctx = app.app_context()
        self.ctx.push()
        db.drop_all()  # just in case
        db.create_all()
        self.client = app.test_client()

def startScheduler():
    db.create_all()
    #create default roles!
    if not db.session.query(models.Role).filter(models.Role.name == "admin").first():
        admin_role = models.Role(name='admin', description='Administrator Role')
        user_role = models.Role(name='user', description='User Role')
        db.session.add(admin_role)
        db.session.add(user_role)
        db.session.commit()

    try:
        import tzlocal

        tz = tzlocal.get_localzone()
        logger.info("local timezone: %s" % tz)
    except:
        tz = None

    if not tz or tz.zone == "local":
        logger.error('Local timezone name could not be determined. Scheduler will display times in UTC for any log'
                 'messages. To resolve this set up /etc/timezone with correct time zone name.')
        tz = pytz.utc
    #in debug mode this is executed twice :(
    #DONT run flask in auto reload mode when testing this!
    scheduler = BackgroundScheduler(logger=sched_logger, timezone=tz)
    scheduler.add_job(notify.task, 'interval', seconds=config.SCAN_INTERVAL, max_instances=1,
                      start_date=datetime.datetime.now(tz) + datetime.timedelta(seconds=2))
    scheduler.start()
    sched = scheduler
    #notify.task()

def run_app():
    from subprocess import Popen
    import sys
    os.chdir(os.path.abspath(os.path.dirname(__file__)))
    path = os.path.join(os.path.abspath(os.path.dirname(__file__)), "manage.py")
    args = [sys.executable, path, "db"]
    if not os.path.exists(os.path.join(config.DATA_DIR, "plexivity.db")):
        from app import db
        db.create_all()
        args.append("stamp")
        args.append("head")
    else:
        args.append("upgrade")

    Popen(args)

    helper.startScheduler()

    if config.USE_SSL:
        helper.generateSSLCert()
        try:
            from OpenSSL import SSL
            context = SSL.Context(SSL.SSLv23_METHOD)
            context.use_privatekey_file(os.path.join(config.DATA_DIR, "plexivity.key"))
            context.use_certificate_file(os.path.join(config.DATA_DIR, "plexivity.crt"))
            app.run(host="0.0.0.0", port=config.PORT, debug=False, ssl_context=context)
        except:
            logger.error("plexivity should use SSL but OpenSSL was not found, starting without SSL")
            app.run(host="0.0.0.0", port=config.PORT, debug=False)
    else:
        app.run(host="0.0.0.0", port=config.PORT, debug=False)

def db(request, app):
    """Create test database tables"""
    _db.drop_all()
    # Create the tables based on the current model
    _db.create_all()

    user = User.create_test_user()
    TestClient.test_user = user
    app.test_client_class = TestClient
    app.response_class = TestResponse

    _db.session.commit()

def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()

def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
        Role.insert_roles()
        self.client = self.app.test_client(use_cookies=True)

def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
        Role.insert_roles()
        self.client = self.app.test_client()

def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()

def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
        Role.insert_roles()

def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()

def setUp(self):
        """
        Create the database
        :return:
        """
        db.create_all()
        db.session.commit()

def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()

    # ?????

def setUp(self):
        self.app = create_app('testing')
        self.context = self.app.app_context()
        self.context.push()
        db.create_all()
        Role.insert_roles()

def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()

def setUp(self):
        self.db_fd, self.db_filename = tempfile.mkstemp()
        # FIXME: this isn't actually working.
        app.config['DATABASE_URI'] = 'sqlite:///' + self.db_filename
        self.app = app.test_client()
        db.create_all()

def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
        Role.insert_roles()
        self.client = self.app.test_client(use_cookies=True)

def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
        Role.insert_roles()
        self.client = self.app.test_client()

def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()