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()
python类models()的实例源码
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