def get(self, shake_name):
shake = Shake.get("name=%s and deleted=0", shake_name)
if not shake:
raise tornado.web.HTTPError(404)
current_user = self.get_current_user_object()
invitation, invitation_requests = _invitations(shake, current_user)
#is this user a shake manager?
managers = shake.managers()
is_shake_manager = False
if managers and current_user:
for manager in managers:
if manager.id == current_user.id:
is_shake_manager = True
break
followers = shake.subscribers()
follower_count = shake.subscriber_count()
return self.render("shakes/members.html", shake=shake, invitation=invitation,
managers=shake.managers(), current_user_obj=current_user,
invitation_requests=invitation_requests, shake_editor=shake.owner(),
is_shake_manager=is_shake_manager, followers=followers,
follower_count=follower_count)
python类get()的实例源码
def get(self):
where = "deleted=0 ORDER BY id DESC LIMIT 21"
before_id = self.get_argument('before', None)
if before_id is not None:
where = "id < %d AND %s" % (int(before_id), where)
users = User.where(where)
prev_link = None
next_link = None
if len(users) == 21:
next_link = "?before=%d" % users[-1].id
for user in users[:20]:
files = user.sharedfiles(per_page=1)
user.last_sharedfile = len(files) == 1 and files[0] or None
return self.render("admin/new-users.html", users=users[:20],
previous_link=prev_link, next_link=next_link)
def get(self):
where = "nsfw=1 AND deleted=0 ORDER BY id DESC LIMIT 21"
before_id = self.get_argument('before', None)
if before_id is not None:
where = "id < %d AND %s" % (int(before_id), where)
users = User.where(where)
prev_link = None
next_link = None
if len(users) == 21:
next_link = "?before=%d" % users[-1].id
for user in users[:20]:
files = user.sharedfiles(per_page=1)
user.last_sharedfile = len(files) == 1 and files[0] or None
return self.render("admin/nsfw-users.html", users=users[:20],
previous_link=prev_link, next_link=next_link)
def post(self):
category = ShakeCategory(name=self.get_argument('name', ''), short_name=self.get_argument('short_name', ''))
category.save()
return self.redirect('/admin/shake-categories')
#class CreateShakeSharedFilesHandler(BaseHandler):
# @tornado.web.authenticated
# def get(self):
# user = self.current_user
# if user['name'] in ['admin',]:
# shared_files = Sharedfile.all()
# for sf in shared_files:
# user = User.get('id = %s', sf.user_id)
# user_shake = user.shake()
# ssf = Shakesharedfile.get("shake_id = %s and sharedfile_id = %s", user_shake.id, sf.id)
# if not ssf:
# ssf = Shakesharedfile(shake_id = user_shake.id, sharedfile_id = sf.id)
# ssf.save()
# else:
# shared_file = Sharedfile.get('id=%s', ssf.sharedfile_id)
# ssf.created_at = shared_file.created_at
# ssf.save()
# return self.redirect("/")
def test_atomic_with_delete(flushdb):
for i in range(3):
await User.create(username=f'u{i}')
async with db.atomic():
user = await User.get(User.username == 'u1')
await user.delete_instance()
usernames = [u.username async for u in User.select()]
assert sorted(usernames) == ['u0', 'u2']
async with db.atomic():
async with db.atomic():
user = await User.get(User.username == 'u2')
await user.delete_instance()
usernames = [u.username async for u in User.select()]
assert usernames == ['u0']
def identity(payload):
user_id = payload['identity']
try:
g.user = User.get(user_id, None)
return g.user
except User.DoesNotExist:
return None
def test_get_users_list_not_authenticated__unauthorized(self):
add_user(None, TEST_USER_PSW)
resp = self.app.get(API_ENDPOINT.format('users/'))
assert resp.status_code == UNAUTHORIZED
def load_user(user_id):
"""
Current user loading logic. If the user exists return it, otherwise None.
Args:
user_id (int): Peewee user id
Returns:
models.User: The requested user
"""
try:
return User.get(User.id == user_id)
except User.DoesNotExist:
return None
def get(self):
""" Get all the orders."""
data = Order.json_list(Order.select())
return generate_response(data, OK)
