1 | """ |
||
2 | byceps.services.shop.catalog.service |
||
3 | ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ |
||
4 | |||
5 | :Copyright: 2006-2020 Jochen Kupperschmidt |
||
6 | :License: Modified BSD, see LICENSE for details. |
||
7 | """ |
||
8 | |||
9 | 1 | from typing import List, Optional |
|
10 | |||
11 | 1 | from ....database import db |
|
12 | |||
13 | 1 | from ..article.transfer.models import ArticleNumber |
|
14 | |||
15 | 1 | from .models import ( |
|
16 | Catalog as DbCatalog, |
||
17 | CatalogArticle as DbCatalogArticle, |
||
18 | Collection as DbCollection, |
||
19 | ) |
||
20 | 1 | from .transfer.models import ( |
|
21 | Catalog, |
||
22 | CatalogArticleID, |
||
23 | CatalogID, |
||
24 | Collection, |
||
25 | CollectionID, |
||
26 | ) |
||
27 | |||
28 | |||
29 | # catalog |
||
30 | |||
31 | |||
32 | 1 | def create_catalog(catalog_id: CatalogID, title: str) -> Catalog: |
|
33 | """Create a catalog.""" |
||
34 | catalog = DbCatalog(catalog_id, title) |
||
35 | |||
36 | db.session.add(catalog) |
||
37 | db.session.commit() |
||
38 | |||
39 | return _db_entity_to_catalog(catalog) |
||
40 | |||
41 | |||
42 | 1 | def find_catalog(catalog_id: CatalogID) -> Optional[Catalog]: |
|
43 | """Return the catalog with that ID, or `None` if not found.""" |
||
44 | catalog = _find_db_catalog(catalog_id) |
||
45 | |||
46 | if catalog is None: |
||
47 | return None |
||
48 | |||
49 | return _db_entity_to_catalog(catalog) |
||
50 | |||
51 | |||
52 | 1 | def _find_db_catalog(catalog_id: CatalogID) -> Optional[DbCatalog]: |
|
53 | """Return the catalog database entity with that ID, or `None` if not |
||
54 | found. |
||
55 | """ |
||
56 | return DbCatalog.query.get(catalog_id) |
||
57 | |||
58 | |||
59 | 1 | def get_all_catalogs() -> List[Catalog]: |
|
60 | """Return all catalogs.""" |
||
61 | catalogs = DbCatalog.query.all() |
||
62 | |||
63 | return [_db_entity_to_catalog(catalog) for catalog in catalogs] |
||
64 | |||
65 | |||
66 | 1 | def _db_entity_to_catalog(catalog: DbCatalog) -> Catalog: |
|
67 | return Catalog( |
||
68 | catalog.id, |
||
69 | catalog.title, |
||
70 | ) |
||
71 | |||
72 | |||
73 | # collection |
||
74 | |||
75 | |||
76 | 1 | def create_collection(catalog_id: CatalogID, title: str) -> Collection: |
|
77 | """Create a collection.""" |
||
78 | catalog = _find_db_catalog(catalog_id) |
||
79 | if catalog is None: |
||
80 | raise ValueError(f'Unknown catalog ID "{catalog_id}"') |
||
81 | |||
82 | collection = DbCollection(catalog_id, title) |
||
83 | |||
84 | catalog.collections.append(collection) |
||
85 | db.session.commit() |
||
86 | |||
87 | return _db_entity_to_collection(collection) |
||
88 | |||
89 | |||
90 | 1 | def delete_collection(collection_id: CollectionID) -> None: |
|
91 | """Delete the collection.""" |
||
92 | db.session.query(DbCollection) \ |
||
93 | .filter_by(id=collection_id) \ |
||
94 | .delete() |
||
95 | |||
96 | db.session.commit() |
||
97 | |||
98 | |||
99 | 1 | def get_collections_for_catalog(catalog_id: CatalogID) -> List[Collection]: |
|
100 | """Return the catalog's collections.""" |
||
101 | rows = DbCollection.query \ |
||
102 | .filter_by(catalog_id=catalog_id) \ |
||
103 | .order_by(DbCollection.position) \ |
||
104 | .all() |
||
105 | |||
106 | return [_db_entity_to_collection(collection) for collection in collections] |
||
0 ignored issues
–
show
Comprehensibility
Best Practice
introduced
by
Loading history...
|
|||
107 | |||
108 | |||
109 | 1 | def _db_entity_to_collection(collection: DbCollection) -> Collection: |
|
110 | return Collection( |
||
111 | collection.id, |
||
112 | collection.catalog_id, |
||
113 | collection.title, |
||
114 | collection.position, |
||
115 | [], |
||
116 | ) |
||
117 | |||
118 | |||
119 | # article assignment |
||
120 | |||
121 | |||
122 | 1 | def add_article_to_collection( |
|
123 | article_number: ArticleNumber, collection_id: CollectionID |
||
124 | ) -> CatalogArticleID: |
||
125 | """Add article to collection.""" |
||
126 | collection = DbCollection.query.get(collection_id) |
||
127 | if collection is None: |
||
128 | raise ValueError(f'Unknown collection ID "{collection_id}"') |
||
129 | |||
130 | catalog_article = DbCatalogArticle(collection_id, article_number) |
||
131 | |||
132 | collection.catalog_articles.append(catalog_article) |
||
133 | db.session.commit() |
||
134 | |||
135 | return catalog_article.id |
||
136 | |||
137 | |||
138 | 1 | def remove_article_from_collection( |
|
139 | catalog_article_id: CatalogArticleID, |
||
140 | ) -> None: |
||
141 | """Remove article from collection.""" |
||
142 | db.session.query(DbCatalogArticle) \ |
||
143 | .filter_by(id=catalog_article_id) \ |
||
144 | .delete() |
||
145 | |||
146 | db.session.commit() |
||
147 |