Python app 模块，models() 实例源码

def run(self):
        mutex.acquire()
        while True:

            Services = models.Service.query.all()
            for c in Services:
                if time.time() - float(c.createdtime) > 600:
                    removeServices(c.serviceid)
                    print "Stopped %s" % c.serviceid
            time.sleep(10)

        mutex.release()

def shell(ipython):
    """Runs a Python shell with QPD context"""
    import code
    import readline
    import rlcompleter
    banner_msg = (
        "Welcome to QPD interactive shell\n"
        "\tAuto imported: app, db, models, views, admin"
    )
    _vars = globals()
    _vars.update(locals())
    _vars.update(dict(app=app, db=db, models=models, views=views, admin=admin))
    readline.set_completer(rlcompleter.Completer(_vars).complete)
    readline.parse_and_bind("tab: complete")
    try:
        if ipython is True:
            from IPython import start_ipython
            from traitlets.config import Config
            c = Config()
            c.TerminalInteractiveShell.banner2 = banner_msg
            start_ipython(argv=[], user_ns=_vars, config=c)
        else:
            raise ImportError
    except ImportError:
        shell = code.InteractiveConsole(_vars)
        shell.interact(banner=banner_msg)

def new(username, email, password):
        user = User()
        user.username = username
        user.email = email
        user.password = User.generate_password_hash(password)
        user.avatar = avatar.init_avatar()
        db.session.add(user)
        db.session.commit()
        UserProfile = app.user.models.UserProfile
        profile = UserProfile()
        profile.user_id = user.id
        db.session.add(profile)
        return user

def get_articles(self, page, topic=None):
        """ ????????? """
        Article = app.article.models.Article
        per_page = current_app.config["USER_PAGE"]
        articles = db.session.query(Article).filter_by(user_id=self.id)
        if not current_user.is_authenticated or current_user.id!=self.id:
            articles = articles.filter_by(access="public")
        if topic:
            ArticleTopic = app.article.models.ArticleTopic
            Topic = app.home.models.Topic
            articles = articles.filter(Article.topics.any(ArticleTopic.topic.has(
                Topic.name==topic.lower())))
        articles = articles.order_by(Article.timestamp.desc())
        articles = paginate(articles, page, per_page=per_page, error_out=False)
        return articles

def prefix_autosearch(name, page, per_page):
        """ ??????? """
        Article = app.article.models.Article
        ArticleTopic = app.article.models.ArticleTopic
        topics = db.session.query(Topic).outerjoin(ArticleTopic, Article).filter(
                Topic.name.ilike(u"{0}%".format(name))).filter(Article.access=="public").group_by(
                Topic).order_by(func.count(ArticleTopic.topic_id).desc())
        return paginate(topics, page, per_page = per_page, error_out = False)

def get_user_article_topics(user_id, page):
        per_page = current_app.config["USER_TOPIC_PAGE"]
        ArticleTopic = app.article.models.ArticleTopic
        Article = app.article.models.Article
        topics = db.session.query(Topic,
            func.count(ArticleTopic.topic_id).label("count"))\
            .outerjoin(ArticleTopic, Article).group_by(Topic)
        if not current_user.is_authenticated or current_user.id!=user_id:
            topics = topics.filter(db.and_(Article.user_id==user_id, 
                Article.access=="public"))
        else:
            topics = topics.filter(Article.user_id==user_id)
        topics = topics.order_by(func.count(ArticleTopic.topic_id).desc())
        topics = paginate(topics, page, per_page=per_page, error_out=False)
        return topics

def get_article_topics(num):
        """ ????????? 
        @param num: ??????
        """
        ArticleTopic = app.article.models.ArticleTopic
        Article = app.article.models.Article
        topics = db.session.query(Topic)
        topics = topics.outerjoin(ArticleTopic).filter(
                ArticleTopic.article.has(Article.access=="public")
                ).group_by(Topic).order_by(
                func.count(ArticleTopic.topic_id).desc())
        return topics.limit(num).all(), topics.count()

def count_article(self):
        Article = app.article.models.Article
        return self.articles.outerjoin(Article).filter(Article.access=="public").count()

def create_app(config_name):
    """
    Given a configuration name, loads the correct 
    configuration from the config.py
    :param config_name: The configuration name to load the configuration
    :return: The app to be initialized
    """
    app = Flask(__name__, instance_relative_config=True)
    app.config.from_object(app_config[config_name])
    app.config.from_pyfile('config.py')
    db.init_app(app)

    """
    Configurations of the flask-login, in which, if user tries to access a 
    page that they are not authorized to, it will redirect to the specific
    view and display the message below on the route auth.login
    """
    login_manager.init_app(app)
    login_manager.login_message = "You must be logged in to access this page"
    login_manager.login_view = "auth.login"

    """
    Migrations setting up.
    This object "migrate" will allow us to run migrations using Flask-Migrate.
    We also have imported the models from the app package.
    """
    migrate = Migrate(app, db)
    Bootstrap(app)

    """
    Import the models to be used in the application
    """
    from app import models

    """
    Configuring the blueprints of each package on the app
    """
    from .admin import admin as admin_blueprint
    # This url_prefix means that all the views for this blueprint will be
    # accessed in the browser with the url prefix admin.
    app.register_blueprint(admin_blueprint, url_prefix='/admin')

    from .auth import auth as auth_blueprint
    app.register_blueprint(auth_blueprint)

    from .home import home as home_blueprint
    app.register_blueprint(home_blueprint)

    return app