~homeworkprod/byceps

ref: bcef32cd2dc329bcf81057794883008f5c040ac8 byceps/byceps/services/shop/catalog/service.py -rw-r--r-- 3.5 KiB
bcef32cd — Jochen Kupperschmidt Work around Jinja 3.0.0 bug with `for` inside of `set` block 1 year, 7 days ago
                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
"""
byceps.services.shop.catalog.service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

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

from __future__ import annotations
from typing import Optional

from ....database import db

from ..article.transfer.models import ArticleNumber

from .dbmodels import (
    Catalog as DbCatalog,
    CatalogArticle as DbCatalogArticle,
    Collection as DbCollection,
)
from .transfer.models import (
    Catalog,
    CatalogArticleID,
    CatalogID,
    Collection,
    CollectionID,
)


# catalog


def create_catalog(catalog_id: CatalogID, title: str) -> Catalog:
    """Create a catalog."""
    catalog = DbCatalog(catalog_id, title)

    db.session.add(catalog)
    db.session.commit()

    return _db_entity_to_catalog(catalog)


def find_catalog(catalog_id: CatalogID) -> Optional[Catalog]:
    """Return the catalog with that ID, or `None` if not found."""
    catalog = _find_db_catalog(catalog_id)

    if catalog is None:
        return None

    return _db_entity_to_catalog(catalog)


def _find_db_catalog(catalog_id: CatalogID) -> Optional[DbCatalog]:
    """Return the catalog database entity with that ID, or `None` if not
    found.
    """
    return DbCatalog.query.get(catalog_id)


def get_all_catalogs() -> list[Catalog]:
    """Return all catalogs."""
    catalogs = DbCatalog.query.all()

    return [_db_entity_to_catalog(catalog) for catalog in catalogs]


def _db_entity_to_catalog(catalog: DbCatalog) -> Catalog:
    return Catalog(
        catalog.id,
        catalog.title,
    )


# collection


def create_collection(catalog_id: CatalogID, title: str) -> Collection:
    """Create a collection."""
    catalog = _find_db_catalog(catalog_id)
    if catalog is None:
        raise ValueError(f'Unknown catalog ID "{catalog_id}"')

    collection = DbCollection(catalog_id, title)

    catalog.collections.append(collection)
    db.session.commit()

    return _db_entity_to_collection(collection)


def delete_collection(collection_id: CollectionID) -> None:
    """Delete the collection."""
    db.session.query(DbCollection) \
        .filter_by(id=collection_id) \
        .delete()

    db.session.commit()


def get_collections_for_catalog(catalog_id: CatalogID) -> list[Collection]:
    """Return the catalog's collections."""
    collections = DbCollection.query \
        .filter_by(catalog_id=catalog_id) \
        .order_by(DbCollection.position) \
        .all()

    return [_db_entity_to_collection(collection) for collection in collections]


def _db_entity_to_collection(collection: DbCollection) -> Collection:
    return Collection(
        collection.id,
        collection.catalog_id,
        collection.title,
        collection.position,
        [],
    )


# article assignment


def add_article_to_collection(
    article_number: ArticleNumber, collection_id: CollectionID
) -> CatalogArticleID:
    """Add article to collection."""
    collection = DbCollection.query.get(collection_id)
    if collection is None:
        raise ValueError(f'Unknown collection ID "{collection_id}"')

    catalog_article = DbCatalogArticle(collection_id, article_number)

    collection.catalog_articles.append(catalog_article)
    db.session.commit()

    return catalog_article.id


def remove_article_from_collection(
    catalog_article_id: CatalogArticleID,
) -> None:
    """Remove article from collection."""
    db.session.query(DbCatalogArticle) \
        .filter_by(id=catalog_article_id) \
        .delete()

    db.session.commit()