forked from google/python-spanner-orm
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmetadata.py
More file actions
106 lines (92 loc) · 4.08 KB
/
metadata.py
File metadata and controls
106 lines (92 loc) · 4.08 KB
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
# python3
# Copyright 2019 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# python3
# Copyright 2019 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Hold information about a Model extracted from the class attributes."""
from typing import Any, Dict, Type, Optional
from spanner_orm import error
from spanner_orm import field
from spanner_orm import index
from spanner_orm import registry
from spanner_orm import relationship
class ModelMetadata(object):
"""Hold information about a Model extracted from the class attributes."""
def __init__(self,
table: Optional[str] = None,
fields: Optional[Dict[str, field.Field]] = None,
relations: Optional[Dict[str, relationship.Relationship]] = None,
indexes: Optional[Dict[str, index.Index]] = None,
interleaved: Optional[str] = None,
model_class: Optional[Type[Any]] = None):
self.columns = []
self.fields = dict(fields or {})
self._finalized = False
self.indexes = dict(indexes or {})
self.interleaved = interleaved
self.model_class = model_class
self.primary_keys = []
self.relations = dict(relations or {})
self.table = table or ''
def finalize(self) -> None:
"""Finish generating metadata state.
Some metadata depends on having all configuration data set before it can
be calculated--the primary index, for example, needs all fields to be added
before it can be calculated. This method is called to indicate that all
relevant state has been added and the calculation of the final data should
now happen.
"""
if self._finalized:
raise error.SpannerError('Metadata was already finalized')
sorted_fields = list(sorted(self.fields.values(), key=lambda f: f.position))
if index.Index.PRIMARY_INDEX not in self.indexes:
primary_keys = [f.name for f in sorted_fields if f.primary_key()]
primary_index = index.Index(primary_keys)
primary_index.name = index.Index.PRIMARY_INDEX
self.indexes[index.Index.PRIMARY_INDEX] = primary_index
self.primary_keys = self.indexes[index.Index.PRIMARY_INDEX].columns
self.columns = [f.name for f in sorted_fields]
for _, relation in self.relations.items():
relation.origin = self.model_class
registry.model_registry().register(self.model_class)
self._finalized = True
def add_metadata(self, metadata: 'ModelMetadata') -> None:
self.table = metadata.table or self.table
self.fields.update(metadata.fields)
self.relations.update(metadata.relations)
self.indexes.update(metadata.indexes)
self.interleaved = metadata.interleaved or self.interleaved
def add_field(self, name: str, new_field: field.Field) -> None:
new_field.name = name
new_field.position = len(self.fields)
self.fields[name] = new_field
def add_relation(self, name: str,
new_relation: relationship.Relationship) -> None:
new_relation.name = name
self.relations[name] = new_relation
def add_index(self, name: str, new_index: index.Index) -> None:
new_index.name = name
self.indexes[name] = new_index