byceps.services.seating.seat_import_service.serialize_seat_to_import() - Code Metrics - byceps/byceps - Measure and Improve Code Quality continuously with Scrutinizer

serialize_seat_to_import() B
last analyzed 2024-05-21 01:28 UTC

↳ Parent: byceps.services.seating.seat_import_service

Complexity

Conditions

Size

Total Lines	31
Code Lines	23

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	30

Importance

Changes

Metric	Value
cc	5
eloc	23
nop	8
dl	0
loc	31
ccs	0
cts	11
cp	0
crap	30
rs	8.8613
c	0
b	0
f	0

How to fix Many Parameters

"""
byceps.services.seating.seat_import_service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

:Copyright: 2014-2024 Jochen Kupperschmidt
:License: Revised BSD (see `LICENSE` file for details)
"""

from __future__ import annotations

from collections.abc import Iterable, Iterator
import json

from pydantic import ValidationError

from byceps.services.party.models import PartyID
from byceps.services.ticketing import ticket_category_service
from byceps.services.ticketing.models.ticket import TicketCategoryID
from byceps.util.result import Err, Ok, Result

from . import seat_group_service, seat_service, seating_area_service
from .models import Seat, SeatingAreaID, SeatToImport, SerializableSeatToImport


def serialize_seat_to_import(
    area_title: str,
    coord_x: int,
    coord_y: int,
    category_title: str,
    rotation: int | None = None,
    label: str | None = None,
    type_: str | None = None,
    group_title: str | None = None,
) -> str:
    """Serialize a seat to JSON so it can be imported."""
    model = SerializableSeatToImport(
        area_title=area_title,
        coord_x=coord_x,
        coord_y=coord_y,
        category_title=category_title,
    )

    if rotation is not None:
        model.rotation = rotation

    if label is not None:
        model.label = label

    if type_ is not None:
        model.type_ = type_

    if group_title is not None:
        model.group_title = group_title

    return model.json(exclude_unset=True)


def load_seats_from_json_lines(
    party_id: PartyID, lines: Iterable[str]
) -> Iterator[tuple[int, Result[SeatToImport, str]]]:
    parser = _create_parser(party_id)
    return parser.parse_lines(lines)


def _create_parser(party_id: PartyID) -> _SeatsImportParser:
    """Create a parser, populated with party-specific data."""
    area_ids_by_title = _get_area_ids_by_title(party_id)
    category_ids_by_title = _get_category_ids_by_title(party_id)
    seat_group_titles = _get_seat_group_titles(party_id)

    return _SeatsImportParser(
        area_ids_by_title, category_ids_by_title, seat_group_titles
    )


def _get_area_ids_by_title(party_id: PartyID) -> dict[str, SeatingAreaID]:
    """Get the party's seating areas as a mapping from title to ID."""
    areas = seating_area_service.get_areas_for_party(party_id)
    return {area.title: area.id for area in areas}


def _get_category_ids_by_title(
    party_id: PartyID,
) -> dict[str, TicketCategoryID]:
    """Get the party's ticket categories as a mapping from title to ID."""
    categories = ticket_category_service.get_categories_for_party(party_id)
    return {category.title: category.id for category in categories}


def _get_seat_group_titles(party_id: PartyID) -> set[str]:
    """Get the party's seat groups' titles."""
    groups = seat_group_service.get_all_seat_groups_for_party(party_id)
    return {group.title for group in groups}


class _SeatsImportParser:
    """Parse JSON Lines records into importable seat objects."""

    def __init__(
        self,
        area_ids_by_title: dict[str, SeatingAreaID],
        category_ids_by_title: dict[str, TicketCategoryID],
        seat_group_titles: set[str],
    ) -> None:
        self._area_ids_by_title = area_ids_by_title
        self._category_ids_by_title = category_ids_by_title
        self._seat_group_titles = seat_group_titles

    def parse_lines(
        self, lines: Iterable[str]
    ) -> Iterator[tuple[int, Result[SeatToImport, str]]]:
        """Parse JSON lines into importable seat objects."""
        for line_number, line in enumerate(lines, start=1):
            parse_result = self._parse_line(line.rstrip())
            yield line_number, parse_result

    def _parse_line(self, line: str) -> Result[SeatToImport, str]:
        """Parse a JSON line into an importable seat object."""
        parse_result = _parse_seat_json(line)
        return parse_result.and_then(self._assemble_seat_to_import)

    def _assemble_seat_to_import(
        self, parsed_seat: SerializableSeatToImport
    ) -> Result[SeatToImport, str]:
        """Build seat object to import by setting area and category IDs."""
        area_title = parsed_seat.area_title
        area_id = self._area_ids_by_title.get(area_title)
        if area_id is None:
            return Err(f'Unknown area title "{area_title}"')

        category_title = parsed_seat.category_title
        category_id = self._category_ids_by_title.get(category_title)
        if category_id is None:
            return Err(f'Unknown category title "{category_title}"')

        group_title = parsed_seat.group_title
        if group_title in self._seat_group_titles:
            return Err(f'Seat group with title "{group_title}" already exists')

        assembled_seat = SeatToImport(
            area_id=area_id,
            coord_x=parsed_seat.coord_x,
            coord_y=parsed_seat.coord_y,
            category_id=category_id,
            rotation=parsed_seat.rotation,
            label=parsed_seat.label,
            type_=parsed_seat.type_,
            group_title=parsed_seat.group_title,
        )
        return Ok(assembled_seat)


def _parse_seat_json(json_data: str) -> Result[SerializableSeatToImport, str]:
    """Parse a JSON object into a seat import object."""
    try:
        data_dict = json.loads(json_data)
    except json.decoder.JSONDecodeError as e:
        return Err(f'Could not parse JSON: {e}')

    try:
        seat_to_import = SerializableSeatToImport.model_validate(data_dict)
        return Ok(seat_to_import)
    except ValidationError as e:
        return Err(str(e))


def import_seat(seat_to_import: SeatToImport) -> Result[Seat, str]:
    """Import a seat."""
    try:
        imported_seat = seat_service.create_seat(
            seat_to_import.area_id,
            seat_to_import.coord_x,
            seat_to_import.coord_y,
            seat_to_import.category_id,
            rotation=seat_to_import.rotation,
            label=seat_to_import.label,
            type_=seat_to_import.type_,
        )
        return Ok(imported_seat)
    except Exception as e:
        return Err(str(e))


1			"""
2			byceps.services.seating.seat_import_service
3			~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5			:Copyright: 2014-2024 Jochen Kupperschmidt
6			:License: Revised BSD (see `LICENSE` file for details)
7			"""
8
9			from __future__ import annotations
10
11			from collections.abc import Iterable, Iterator
12			import json
13
14			from pydantic import ValidationError
15
16			from byceps.services.party.models import PartyID
17			from byceps.services.ticketing import ticket_category_service
18			from byceps.services.ticketing.models.ticket import TicketCategoryID
19			from byceps.util.result import Err, Ok, Result
20
21			from . import seat_group_service, seat_service, seating_area_service
22			from .models import Seat, SeatingAreaID, SeatToImport, SerializableSeatToImport
23
24
25			def serialize_seat_to_import(
26			area_title: str,
27			coord_x: int,
28			coord_y: int,
29			category_title: str,
30			rotation: int \| None = None,
31			label: str \| None = None,
32			type_: str \| None = None,
33			group_title: str \| None = None,
34			) -> str:
35			"""Serialize a seat to JSON so it can be imported."""
36			model = SerializableSeatToImport(
37			area_title=area_title,
38			coord_x=coord_x,
39			coord_y=coord_y,
40			category_title=category_title,
41			)
42
43			if rotation is not None:
44			model.rotation = rotation
45
46			if label is not None:
47			model.label = label
48
49			if type_ is not None:
50			model.type_ = type_
51
52			if group_title is not None:
53			model.group_title = group_title
54
55			return model.json(exclude_unset=True)
56
57
58			def load_seats_from_json_lines(
59			party_id: PartyID, lines: Iterable[str]
60			) -> Iterator[tuple[int, Result[SeatToImport, str]]]:
61			parser = _create_parser(party_id)
62			return parser.parse_lines(lines)
63
64
65			def _create_parser(party_id: PartyID) -> _SeatsImportParser:
66			"""Create a parser, populated with party-specific data."""
67			area_ids_by_title = _get_area_ids_by_title(party_id)
68			category_ids_by_title = _get_category_ids_by_title(party_id)
69			seat_group_titles = _get_seat_group_titles(party_id)
70
71			return _SeatsImportParser(
72			area_ids_by_title, category_ids_by_title, seat_group_titles
73			)
74
75
76			def _get_area_ids_by_title(party_id: PartyID) -> dict[str, SeatingAreaID]:
77			"""Get the party's seating areas as a mapping from title to ID."""
78			areas = seating_area_service.get_areas_for_party(party_id)
79			return {area.title: area.id for area in areas}
80
81
82			def _get_category_ids_by_title(
83			party_id: PartyID,
84			) -> dict[str, TicketCategoryID]:
85			"""Get the party's ticket categories as a mapping from title to ID."""
86			categories = ticket_category_service.get_categories_for_party(party_id)
87			return {category.title: category.id for category in categories}
88
89
90			def _get_seat_group_titles(party_id: PartyID) -> set[str]:
91			"""Get the party's seat groups' titles."""
92			groups = seat_group_service.get_all_seat_groups_for_party(party_id)
93			return {group.title for group in groups}
94
95
96			class _SeatsImportParser:
97			"""Parse JSON Lines records into importable seat objects."""
98
99			def __init__(
100			self,
101			area_ids_by_title: dict[str, SeatingAreaID],
102			category_ids_by_title: dict[str, TicketCategoryID],
103			seat_group_titles: set[str],
104			) -> None:
105			self._area_ids_by_title = area_ids_by_title
106			self._category_ids_by_title = category_ids_by_title
107			self._seat_group_titles = seat_group_titles
108
109			def parse_lines(
110			self, lines: Iterable[str]
111			) -> Iterator[tuple[int, Result[SeatToImport, str]]]:
112			"""Parse JSON lines into importable seat objects."""
113			for line_number, line in enumerate(lines, start=1):
114			parse_result = self._parse_line(line.rstrip())
115			yield line_number, parse_result
116
117			def _parse_line(self, line: str) -> Result[SeatToImport, str]:
118			"""Parse a JSON line into an importable seat object."""
119			parse_result = _parse_seat_json(line)
120			return parse_result.and_then(self._assemble_seat_to_import)
121
122			def _assemble_seat_to_import(
123			self, parsed_seat: SerializableSeatToImport
124			) -> Result[SeatToImport, str]:
125			"""Build seat object to import by setting area and category IDs."""
126			area_title = parsed_seat.area_title
127			area_id = self._area_ids_by_title.get(area_title)
128			if area_id is None:
129			return Err(f'Unknown area title "{area_title}"')
130
131			category_title = parsed_seat.category_title
132			category_id = self._category_ids_by_title.get(category_title)
133			if category_id is None:
134			return Err(f'Unknown category title "{category_title}"')
135
136			group_title = parsed_seat.group_title
137			if group_title in self._seat_group_titles:
138			return Err(f'Seat group with title "{group_title}" already exists')
139
140			assembled_seat = SeatToImport(
141			area_id=area_id,
142			coord_x=parsed_seat.coord_x,
143			coord_y=parsed_seat.coord_y,
144			category_id=category_id,
145			rotation=parsed_seat.rotation,
146			label=parsed_seat.label,
147			type_=parsed_seat.type_,
148			group_title=parsed_seat.group_title,
149			)
150			return Ok(assembled_seat)
151
152
153			def _parse_seat_json(json_data: str) -> Result[SerializableSeatToImport, str]:
154			"""Parse a JSON object into a seat import object."""
155			try:
156			data_dict = json.loads(json_data)
157			except json.decoder.JSONDecodeError as e:
158			return Err(f'Could not parse JSON: {e}')
159
160			try:
161			seat_to_import = SerializableSeatToImport.model_validate(data_dict)
162			return Ok(seat_to_import)
163			except ValidationError as e:
164			return Err(str(e))
165
166
167			def import_seat(seat_to_import: SeatToImport) -> Result[Seat, str]:
168			"""Import a seat."""
169			try:
170			imported_seat = seat_service.create_seat(
171			seat_to_import.area_id,
172			seat_to_import.coord_x,
173			seat_to_import.coord_y,
174			seat_to_import.category_id,
175			rotation=seat_to_import.rotation,
176			label=seat_to_import.label,
177			type_=seat_to_import.type_,
178			)
179			return Ok(imported_seat)
180			except Exception as e:
181			return Err(str(e))
182

byceps / byceps

serialize_seat_to_import() B last analyzed 2024-05-21 01:28 UTC

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Many Parameters

Many Parameters

Duplication Side-by-Side

Filter issues like

serialize_seat_to_import() B
last analyzed 2024-05-21 01:28 UTC