~homeworkprod/byceps

2d3a7010a02cdadadec65fb21691f75db7132215 — Jochen Kupperschmidt 1 year, 2 months ago d0a9c06
Introduce user service function `get_user`
M byceps/announce/irc/board.py => byceps/announce/irc/board.py +20 -20
@@ 45,7 45,7 @@ def announce_board_topic_created(event: BoardTopicCreated) -> None:
    channels = [CHANNEL_ORGA_LOG, CHANNEL_PUBLIC]

    topic = board_topic_query_service.find_topic_by_id(event.topic_id)
    topic_creator = user_service.find_user(topic.creator_id)
    topic_creator = user_service.get_user(topic.creator_id)
    board_label = _get_board_label(topic)

    text = (


@@ 66,9 66,9 @@ def announce_board_topic_hidden(event: BoardTopicHidden) -> None:
    channels = [CHANNEL_ORGA_LOG]

    topic = board_topic_query_service.find_topic_by_id(event.topic_id)
    topic_creator = user_service.find_user(topic.creator_id)
    topic_creator = user_service.get_user(topic.creator_id)
    board_label = _get_board_label(topic)
    moderator = user_service.find_user(event.moderator_id)
    moderator = user_service.get_user(event.moderator_id)

    text = (
        f'{moderator.screen_name} hat im {board_label} '


@@ 91,9 91,9 @@ def announce_board_topic_unhidden(event: BoardTopicUnhidden) -> None:
    channels = [CHANNEL_ORGA_LOG]

    topic = board_topic_query_service.find_topic_by_id(event.topic_id)
    topic_creator = user_service.find_user(topic.creator_id)
    topic_creator = user_service.get_user(topic.creator_id)
    board_label = _get_board_label(topic)
    moderator = user_service.find_user(event.moderator_id)
    moderator = user_service.get_user(event.moderator_id)

    text = (
        f'{moderator.screen_name} hat im {board_label} '


@@ 114,9 114,9 @@ def announce_board_topic_locked(event: BoardTopicLocked) -> None:
    channels = [CHANNEL_ORGA_LOG]

    topic = board_topic_query_service.find_topic_by_id(event.topic_id)
    topic_creator = user_service.find_user(topic.creator_id)
    topic_creator = user_service.get_user(topic.creator_id)
    board_label = _get_board_label(topic)
    moderator = user_service.find_user(event.moderator_id)
    moderator = user_service.get_user(event.moderator_id)

    text = (
        f'{moderator.screen_name} hat im {board_label} '


@@ 139,9 139,9 @@ def announce_board_topic_unlocked(event: BoardTopicUnlocked) -> None:
    channels = [CHANNEL_ORGA_LOG]

    topic = board_topic_query_service.find_topic_by_id(event.topic_id)
    topic_creator = user_service.find_user(topic.creator_id)
    topic_creator = user_service.get_user(topic.creator_id)
    board_label = _get_board_label(topic)
    moderator = user_service.find_user(event.moderator_id)
    moderator = user_service.get_user(event.moderator_id)

    text = (
        f'{moderator.screen_name} hat im {board_label} '


@@ 162,9 162,9 @@ def announce_board_topic_pinned(event: BoardTopicPinned) -> None:
    channels = [CHANNEL_ORGA_LOG]

    topic = board_topic_query_service.find_topic_by_id(event.topic_id)
    topic_creator = user_service.find_user(topic.creator_id)
    topic_creator = user_service.get_user(topic.creator_id)
    board_label = _get_board_label(topic)
    moderator = user_service.find_user(event.moderator_id)
    moderator = user_service.get_user(event.moderator_id)

    text = (
        f'{moderator.screen_name} hat im {board_label} '


@@ 187,9 187,9 @@ def announce_board_topic_unpinned(event: BoardTopicUnpinned) -> None:
    channels = [CHANNEL_ORGA_LOG]

    topic = board_topic_query_service.find_topic_by_id(event.topic_id)
    topic_creator = user_service.find_user(topic.creator_id)
    topic_creator = user_service.get_user(topic.creator_id)
    board_label = _get_board_label(topic)
    moderator = user_service.find_user(event.moderator_id)
    moderator = user_service.get_user(event.moderator_id)

    text = (
        f'{moderator.screen_name} hat im {board_label} '


@@ 210,9 210,9 @@ def announce_board_topic_moved(event: BoardTopicMoved) -> None:
    channels = [CHANNEL_ORGA_LOG]

    topic = board_topic_query_service.find_topic_by_id(event.topic_id)
    topic_creator = user_service.find_user(topic.creator_id)
    topic_creator = user_service.get_user(topic.creator_id)
    board_label = _get_board_label(topic)
    moderator = user_service.find_user(event.moderator_id)
    moderator = user_service.get_user(event.moderator_id)

    text = (
        f'{moderator.screen_name} hat im {board_label} '


@@ 236,7 236,7 @@ def announce_board_posting_created(event: BoardPostingCreated) -> None:
    channels = [CHANNEL_ORGA_LOG, CHANNEL_PUBLIC]

    posting = board_posting_query_service.find_posting_by_id(event.posting_id)
    posting_creator = user_service.find_user(posting.creator_id)
    posting_creator = user_service.get_user(posting.creator_id)
    board_label = _get_board_label(posting.topic)

    if event.topic_muted:


@@ 262,9 262,9 @@ def announce_board_posting_hidden(event: BoardPostingHidden) -> None:
    channels = [CHANNEL_ORGA_LOG]

    posting = board_posting_query_service.find_posting_by_id(event.posting_id)
    posting_creator = user_service.find_user(posting.creator_id)
    posting_creator = user_service.get_user(posting.creator_id)
    board_label = _get_board_label(posting.topic)
    moderator = user_service.find_user(event.moderator_id)
    moderator = user_service.get_user(event.moderator_id)

    text = (
        f'{moderator.screen_name} hat im {board_label} '


@@ 287,9 287,9 @@ def announce_board_posting_unhidden(event: BoardPostingUnhidden) -> None:
    channels = [CHANNEL_ORGA_LOG]

    posting = board_posting_query_service.find_posting_by_id(event.posting_id)
    posting_creator = user_service.find_user(posting.creator_id)
    posting_creator = user_service.get_user(posting.creator_id)
    board_label = _get_board_label(posting.topic)
    moderator = user_service.find_user(event.moderator_id)
    moderator = user_service.get_user(event.moderator_id)

    text = (
        f'{moderator.screen_name} hat im {board_label} '

M byceps/announce/irc/user.py => byceps/announce/irc/user.py +13 -13
@@ 34,10 34,10 @@ def announce_user_account_created(event: UserAccountCreated) -> None:
    """Announce that a user account has been created."""
    channels = [CHANNEL_ORGA_LOG]

    user = user_service.find_user(event.user_id)
    user = user_service.get_user(event.user_id)

    if event.initiator_id is not None:
        initiator = user_service.find_user(event.initiator_id)
        initiator = user_service.get_user(event.initiator_id)
        initiator_label = initiator.screen_name
    else:
        initiator_label = 'Jemand'


@@ 61,7 61,7 @@ def announce_user_screen_name_changed(event: UserScreenNameChanged) -> None:
    """Announce that a user's screen name has been changed."""
    channels = [CHANNEL_ORGA_LOG]

    initiator = user_service.find_user(event.initiator_id)
    initiator = user_service.get_user(event.initiator_id)

    text = (
        f'{initiator.screen_name} hat das Benutzerkonto '


@@ 84,8 84,8 @@ def announce_user_email_address_invalidated(
    """Announce that a user's email address has been invalidated."""
    channels = [CHANNEL_ORGA_LOG]

    user = user_service.find_user(event.user_id)
    initiator = user_service.find_user(event.initiator_id)
    user = user_service.get_user(event.user_id)
    initiator = user_service.get_user(event.initiator_id)

    text = (
        f'{initiator.screen_name} hat die E-Mail-Adresse '


@@ 106,8 106,8 @@ def announce_user_details_updated_changed(event: UserDetailsUpdated) -> None:
    """Announce that a user's details have been changed."""
    channels = [CHANNEL_ORGA_LOG]

    user = user_service.find_user(event.user_id)
    initiator = user_service.find_user(event.initiator_id)
    user = user_service.get_user(event.user_id)
    initiator = user_service.get_user(event.initiator_id)

    text = (
        f'{initiator.screen_name} hat die persönlichen Daten '


@@ 126,8 126,8 @@ def announce_user_account_suspended(event: UserAccountSuspended) -> None:
    """Announce that a user account has been suspended."""
    channels = [CHANNEL_ORGA_LOG]

    user = user_service.find_user(event.user_id)
    initiator = user_service.find_user(event.initiator_id)
    user = user_service.get_user(event.user_id)
    initiator = user_service.get_user(event.initiator_id)

    text = (
        f'{initiator.screen_name} hat das Benutzerkonto '


@@ 148,8 148,8 @@ def announce_user_account_unsuspended(event: UserAccountUnsuspended) -> None:
    """Announce that a user account has been unsuspended."""
    channels = [CHANNEL_ORGA_LOG]

    user = user_service.find_user(event.user_id)
    initiator = user_service.find_user(event.initiator_id)
    user = user_service.get_user(event.user_id)
    initiator = user_service.get_user(event.initiator_id)

    text = (
        f'{initiator.screen_name} hat das Benutzerkonto '


@@ 168,8 168,8 @@ def announce_user_account_deleted(event: UserAccountDeleted) -> None:
    """Announce that a user account has been created."""
    channels = [CHANNEL_ORGA_LOG]

    user = user_service.find_user(event.user_id)
    initiator = user_service.find_user(event.initiator_id)
    user = user_service.get_user(event.user_id)
    initiator = user_service.get_user(event.initiator_id)

    text = (
        f'{initiator.screen_name} hat das Benutzerkonto '

M byceps/announce/irc/user_badge.py => byceps/announce/irc/user_badge.py +2 -2
@@ 27,11 27,11 @@ def announce_user_badge_awarded(event: UserBadgeAwarded) -> None:
    """Announce that a badge has been awarded to a user."""
    channels = [CHANNEL_ORGA_LOG]

    user = user_service.find_user(event.user_id)
    user = user_service.get_user(event.user_id)
    badge = user_badge_service.find_badge(event.badge_id)

    if event.initiator_id:
        initiator = user_service.find_user(event.initiator_id)
        initiator = user_service.get_user(event.initiator_id)
        initiator_name = initiator.screen_name
    else:
        initiator_name = 'Jemand'

M byceps/blueprints/admin/orga_team/views.py => byceps/blueprints/admin/orga_team/views.py +4 -4
@@ 245,7 245,7 @@ def membership_create(team_id):
    if not form.validate():
        return membership_create_form(team.id, form)

    user = user_service.find_user(form.user_id.data)
    user = user_service.get_user(form.user_id.data)
    duties = form.duties.data.strip()

    membership = orga_team_service.create_membership(team.id, user.id, duties)


@@ 263,7 263,7 @@ def membership_update_form(membership_id, erroneous_form=None):
    """Show form to update a membership."""
    membership = _get_membership_or_404(membership_id)

    user = user_service.find_user(membership.user_id)
    user = user_service.get_user(membership.user_id)
    team = orga_team_service.find_team(membership.orga_team_id)
    party = party_service.get_party(team.party_id)



@@ 291,7 291,7 @@ def membership_update(membership_id):
    """Update a membership."""
    membership = _get_membership_or_404(membership_id)

    user = user_service.find_user(membership.user_id)
    user = user_service.get_user(membership.user_id)
    team = orga_team_service.find_team(membership.orga_team_id)

    teams = orga_team_service.get_teams_for_party(team.party_id)


@@ 321,7 321,7 @@ def membership_remove(membership_id):
    """Remove an organizer from a team."""
    membership = _get_membership_or_404(membership_id)

    user = user_service.find_user(membership.user_id)
    user = user_service.get_user(membership.user_id)
    team = orga_team_service.find_team(membership.orga_team_id)

    orga_team_service.delete_membership(membership.id)

M byceps/blueprints/admin/shop/order/service.py => byceps/blueprints/admin/shop/order/service.py +1 -1
@@ 124,7 124,7 @@ def _get_additional_data_for_badge_awarded(event: OrderEvent) -> OrderEventData:
    badge = user_badge_service.find_badge(badge_id)

    recipient_id = event.data['recipient_id']
    recipient = user_service.find_user(recipient_id)
    recipient = user_service.get_user(recipient_id)

    return {
        'badge_label': badge.label,

M byceps/blueprints/admin/shop/order/views.py => byceps/blueprints/admin/shop/order/views.py +1 -1
@@ 96,7 96,7 @@ def view(order_id):
    if order is None:
        abort(404)

    placed_by = user_service.find_user(order.placed_by_id, include_avatar=True)
    placed_by = user_service.get_user(order.placed_by_id, include_avatar=True)

    shop = shop_service.get_shop(order.shop_id)


M byceps/blueprints/board/views_posting.py => byceps/blueprints/board/views_posting.py +2 -2
@@ 74,7 74,7 @@ def quote_posting_as_bbcode():
        flash_error('Der zu zitierende Beitrag wurde nicht gefunden.')
        return

    creator = user_service.find_user(posting.creator_id)
    creator = user_service.get_user(posting.creator_id)

    return f'[quote author="{creator.screen_name}"]{posting.body}[/quote]'



@@ 219,7 219,7 @@ def posting_moderate_form(posting_id):
    """Show a form to moderate the posting."""
    posting = h.get_posting_or_404(posting_id)

    posting.creator = user_service.find_user(posting.creator_id)
    posting.creator = user_service.get_user(posting.creator_id)

    return {
        'posting': posting,

M byceps/blueprints/board/views_topic.py => byceps/blueprints/board/views_topic.py +1 -1
@@ 252,7 252,7 @@ def topic_moderate_form(topic_id):
    board_id = h.get_board_id()
    topic = h.get_topic_or_404(topic_id)

    topic.creator = user_service.find_user(topic.creator_id)
    topic.creator = user_service.get_user(topic.creator_id)

    categories = board_category_query_service.get_categories_excluding(
        board_id, topic.category_id

M byceps/services/shop/order/email/service.py => byceps/services/shop/order/email/service.py +1 -1
@@ 134,7 134,7 @@ def _get_order_email_data(order_id: OrderID) -> OrderEmailData:

    shop = shop_service.get_shop(order.shop_id)
    orderer_id = order.placed_by_id
    screen_name = user_service.find_user(orderer_id).screen_name
    screen_name = user_service.get_user(orderer_id).screen_name
    email_address = user_service.get_email_address(orderer_id)

    return OrderEmailData(

M byceps/services/shop/order/export/service.py => byceps/services/shop/order/export/service.py +1 -1
@@ 39,7 39,7 @@ def export_order_as_xml(order_id: OrderID) -> Optional[Dict[str, str]]:

def _assemble_context(order: Order) -> Dict[str, Any]:
    """Assemble template context."""
    placed_by = user_service.find_user(order.placed_by_id)
    placed_by = user_service.get_user(order.placed_by_id)
    email_address = user_service.get_email_address(placed_by.id)

    now = datetime.utcnow()

M byceps/services/tourney/match_comment_service.py => byceps/services/tourney/match_comment_service.py +5 -7
@@ 29,17 29,17 @@ def find_comment(
        return None

    # creator
    creator = _find_user(comment.created_by_id, party_id)
    creator = _get_user(comment.created_by_id, party_id)

    # last editor
    last_editor = None
    if comment.last_edited_by_id:
        last_editor = _find_user(comment.last_edited_by_id, party_id)
        last_editor = _get_user(comment.last_edited_by_id, party_id)

    # moderator
    moderator = None
    if comment.hidden_by_id:
        moderator = _find_user(comment.hidden_by_id, party_id)
        moderator = _get_user(comment.hidden_by_id, party_id)

    return _db_entity_to_comment(
        comment, creator, last_editor=last_editor, moderator=moderator,


@@ 183,10 183,8 @@ def _get_db_comment(comment_id: MatchCommentID) -> DbMatchComment:
    return comment


def _find_user(
    user_id: UserID, party_id: Optional[PartyID] = None
) -> Optional[User]:
    return user_service.find_user(
def _get_user(user_id: UserID, party_id: Optional[PartyID] = None) -> User:
    return user_service.get_user(
        user_id, include_avatar=True, include_orga_flag_for_party_id=party_id
    )


M byceps/services/user/service.py => byceps/services/user/service.py +23 -1
@@ 84,7 84,7 @@ def find_user(
) -> Optional[User]:
    """Return the user with that ID, or `None` if not found.

    Include avatar URLs if requested.
    Include avatar URL if requested.
    """
    row = _get_user_query(include_avatar, include_orga_flag_for_party_id) \
        .filter(DbUser.id == user_id) \


@@ 96,6 96,28 @@ def find_user(
    return _user_row_to_dto(row)


def get_user(
    user_id: UserID,
    *,
    include_avatar: bool = False,
    include_orga_flag_for_party_id: Optional[PartyID] = None,
) -> Optional[User]:
    """Return the user with that ID, or raise an exception.

    Include avatar URL if requested.
    """
    user = find_user(
        user_id,
        include_avatar=include_avatar,
        include_orga_flag_for_party_id=include_orga_flag_for_party_id,
    )

    if user is None:
        raise ValueError(f"Unknown user ID '{user_id}'")

    return user


def find_users(
    user_ids: Set[UserID],
    *,

M tests/integration/services/user/test_suspend_unsuspend.py => tests/integration/services/user/test_suspend_unsuspend.py +4 -4
@@ 30,7 30,7 @@ def test_suspend(admin_app, cheater, admin_user):

    reason = 'User has been caught cheating.'

    user_before = user_service.find_user(user_id)
    user_before = user_service.get_user(user_id)
    assert not user_before.suspended

    events_before = event_service.get_events_for_user(user_before.id)


@@ 42,7 42,7 @@ def test_suspend(admin_app, cheater, admin_user):

    # -------------------------------- #

    user_after = user_service.find_user(user_id)
    user_after = user_service.get_user(user_id)
    assert user_after.suspended

    events_after = event_service.get_events_for_user(user_after.id)


@@ 63,7 63,7 @@ def test_unsuspend(admin_app, remorseful_user, admin_user):

    reason = 'User showed penitence. Drop the ban.'

    user_before = user_service.find_user(user_id)
    user_before = user_service.get_user(user_id)
    assert user_before.suspended

    events_before = event_service.get_events_for_user(user_before.id)


@@ 75,7 75,7 @@ def test_unsuspend(admin_app, remorseful_user, admin_user):

    # -------------------------------- #

    user_after = user_service.find_user(user_id)
    user_after = user_service.get_user(user_id)
    assert not user_after.suspended

    events_after = event_service.get_events_for_user(user_after.id)