mirror of https://gitlab.com/tildes/tildes.git
pollev
2 months ago
3 changed files with 334 additions and 0 deletions
@ -0,0 +1,215 @@ |
|||
# Copyright (c) 2018 Tildes contributors <code@tildes.net> |
|||
# SPDX-License-Identifier: AGPL-3.0-or-later |
|||
|
|||
"""JSON API endpoints related to users.""" |
|||
|
|||
from pyramid.request import Request |
|||
from pyramid.view import view_config |
|||
from tildes.models.user.user import User |
|||
from tildes.models.comment import Comment |
|||
from tildes.models.pagination import MixedPaginatedResults |
|||
from tildes.models.topic import Topic |
|||
from tildes.views.api.beta.api_utils import ( |
|||
build_error_response, |
|||
get_next_and_prev_link, |
|||
query_apply_pagination, |
|||
) |
|||
from tildes.views.api.beta.comment import comment_to_dict |
|||
from tildes.views.api.beta.topic import topic_to_dict |
|||
|
|||
|
|||
def _user_to_dict(user: User) -> dict: |
|||
"""Convert a User object to a dictionary for JSON serialization.""" |
|||
return { |
|||
"username": user.username, |
|||
"joined_at": user.created_time.isoformat(), |
|||
"bio_rendered_html": user.bio_rendered_html, |
|||
} |
|||
|
|||
|
|||
@view_config(route_name="apibeta.user", openapi=True, renderer="json") |
|||
def get_user(request: Request) -> dict: # noqa |
|||
"""Get a single user with their comment and post history.""" |
|||
username = request.openapi_validated.parameters.path.get("username") |
|||
limit = request.openapi_validated.parameters.query.get("limit", 20) |
|||
before = request.openapi_validated.parameters.query.get("before", None) |
|||
after = request.openapi_validated.parameters.query.get("after", None) |
|||
|
|||
# Maximum number of items to return without history permission |
|||
max_items_no_permission = 20 |
|||
|
|||
try: |
|||
query = request.query(User).include_deleted().filter(User.username == username) |
|||
user = query.one_or_none() |
|||
if not user: |
|||
raise ValueError(f"User with name {username} not found") |
|||
except ValueError as exc: |
|||
return build_error_response(str(exc), field="username") |
|||
|
|||
if not request.has_permission("view_history", user) and ( |
|||
limit > max_items_no_permission or before or after |
|||
): |
|||
return build_error_response( |
|||
f"You do not have permission to view this user's history after " |
|||
f"the first {max_items_no_permission} items. " |
|||
f"Please resubmit your request without pagination parameters. " |
|||
f"If you submit a limit, it must be less " |
|||
f"than or equal to {max_items_no_permission}.", |
|||
status=403, |
|||
field="limit/before/after", |
|||
error_type="AuthorizationError", |
|||
) |
|||
|
|||
result_sets = [] |
|||
# For the main user API endpoint, combine topics and comments |
|||
for type_to_query in [Topic, Comment]: |
|||
query = request.query(type_to_query).filter(type_to_query.user == user) |
|||
try: |
|||
query = query_apply_pagination( |
|||
query, before, after, error_if_no_anchor=True |
|||
) |
|||
except ValueError as exc: |
|||
return build_error_response(str(exc), field="pagination") |
|||
# include removed posts if the viewer has permission |
|||
if request.has_permission("view_removed_posts", user): |
|||
query = query.include_removed() |
|||
query = query.join_all_relationships() |
|||
|
|||
result_sets.append(query.get_page(limit)) |
|||
|
|||
combined_results = MixedPaginatedResults(result_sets) |
|||
|
|||
# Build the JSON history data |
|||
processed_results = [] |
|||
for item in combined_results.results: |
|||
if isinstance(item, Topic): |
|||
processed_results.append(topic_to_dict(item)) |
|||
elif isinstance(item, Comment): |
|||
processed_results.append(comment_to_dict(request, item)) |
|||
|
|||
# Construct the paging next and previous link if there are more topics |
|||
(next_link, prev_link) = get_next_and_prev_link(request, combined_results) |
|||
|
|||
# Construct the final response JSON object |
|||
response = { |
|||
"user": _user_to_dict(user), |
|||
"history": processed_results, |
|||
"pagination": { |
|||
"num_items": len(processed_results), |
|||
"next_link": next_link, |
|||
"prev_link": prev_link, |
|||
}, |
|||
} |
|||
return response |
|||
|
|||
|
|||
@view_config(route_name="apibeta.user_comments", openapi=True, renderer="json") |
|||
def get_user_comments(request: Request) -> dict: |
|||
"""Get comments made by a user.""" |
|||
username = request.openapi_validated.parameters.path.get("username") |
|||
limit = request.openapi_validated.parameters.query.get("limit", 50) |
|||
before = request.openapi_validated.parameters.query.get("before", None) |
|||
after = request.openapi_validated.parameters.query.get("after", None) |
|||
|
|||
try: |
|||
query = request.query(User).include_deleted().filter(User.username == username) |
|||
user = query.one_or_none() |
|||
if not user: |
|||
raise ValueError(f"User with name {username} not found") |
|||
except ValueError as exc: |
|||
return build_error_response(str(exc), field="username") |
|||
|
|||
if not request.has_permission("view_history", user): |
|||
return build_error_response( |
|||
"You do not have permission to view this user's comments.", |
|||
status=403, |
|||
field="N/A", |
|||
error_type="AuthorizationError", |
|||
) |
|||
|
|||
query = request.query(Comment).filter(Comment.user == user) |
|||
try: |
|||
query = query_apply_pagination(query, before, after, error_if_no_anchor=False) |
|||
except ValueError as exc: |
|||
return build_error_response(str(exc), field="pagination") |
|||
# include removed posts if the viewer has permission |
|||
if request.has_permission("view_removed_posts", user): |
|||
query = query.include_removed() |
|||
query = query.join_all_relationships() |
|||
|
|||
query_result = query.get_page(limit) |
|||
|
|||
# Build the JSON history data |
|||
processed_comments = [] |
|||
for comment in query_result.results: |
|||
processed_comments.append(comment_to_dict(request, comment)) |
|||
|
|||
# Construct the paging next and previous link if there are more comments |
|||
(next_link, prev_link) = get_next_and_prev_link(request, query_result) |
|||
|
|||
# Construct the final response JSON object |
|||
response = { |
|||
"comments": processed_comments, |
|||
"pagination": { |
|||
"num_items": len(processed_comments), |
|||
"next_link": next_link, |
|||
"prev_link": prev_link, |
|||
}, |
|||
} |
|||
return response |
|||
|
|||
|
|||
@view_config(route_name="apibeta.user_topics", openapi=True, renderer="json") |
|||
def get_user_topics(request: Request) -> dict: |
|||
"""Get topics made by a user.""" |
|||
username = request.openapi_validated.parameters.path.get("username") |
|||
limit = request.openapi_validated.parameters.query.get("limit", 50) |
|||
before = request.openapi_validated.parameters.query.get("before", None) |
|||
after = request.openapi_validated.parameters.query.get("after", None) |
|||
|
|||
try: |
|||
query = request.query(User).include_deleted().filter(User.username == username) |
|||
user = query.one_or_none() |
|||
if not user: |
|||
raise ValueError(f"User with name {username} not found") |
|||
except ValueError as exc: |
|||
return build_error_response(str(exc), field="username") |
|||
|
|||
if not request.has_permission("view_history", user): |
|||
return build_error_response( |
|||
"You do not have permission to view this user's topics.", |
|||
status=403, |
|||
field="N/A", |
|||
error_type="AuthorizationError", |
|||
) |
|||
|
|||
query = request.query(Topic).filter(Topic.user == user) |
|||
try: |
|||
query = query_apply_pagination(query, before, after, error_if_no_anchor=False) |
|||
except ValueError as exc: |
|||
return build_error_response(str(exc), field="pagination") |
|||
# include removed posts if the viewer has permission |
|||
if request.has_permission("view_removed_posts", user): |
|||
query = query.include_removed() |
|||
query = query.join_all_relationships() |
|||
|
|||
query_result = query.get_page(limit) |
|||
|
|||
# Build the JSON history data |
|||
processed_topics = [] |
|||
for topic in query_result.results: |
|||
processed_topics.append(topic_to_dict(topic)) |
|||
|
|||
# Construct the paging next and previous link if there are more topics |
|||
(next_link, prev_link) = get_next_and_prev_link(request, query_result) |
|||
|
|||
# Construct the final response JSON object |
|||
response = { |
|||
"topics": processed_topics, |
|||
"pagination": { |
|||
"num_items": len(processed_topics), |
|||
"next_link": next_link, |
|||
"prev_link": prev_link, |
|||
}, |
|||
} |
|||
return response |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue