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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
|
#!/usr/bin/env python3
##
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you 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.
"""Contains the Name classes."""
from typing import TYPE_CHECKING, Dict, Optional
from avro.constants import PRIMITIVE_TYPES
if TYPE_CHECKING:
from avro.schema import NamedSchema
import re
import warnings
import avro.errors
# The name portion of a fullname, record field names, and enum symbols must:
# start with [A-Za-z_]
# subsequently contain only [A-Za-z0-9_]
_BASE_NAME_PATTERN = re.compile(r"(?:^|\.)[A-Za-z_][A-Za-z0-9_]*$")
def validate_basename(basename: str) -> None:
"""Raise InvalidName if the given basename is not a valid name."""
if not _BASE_NAME_PATTERN.search(basename):
raise avro.errors.InvalidName(f"{basename!s} is not a valid Avro name because it does not match the pattern {_BASE_NAME_PATTERN.pattern!s}")
def _validate_fullname(fullname: str) -> None:
for name in fullname.split("."):
validate_basename(name)
class Name:
"""Class to describe Avro name."""
_full: Optional[str] = None
_name: Optional[str] = None
def __init__(
self, name_attr: Optional[str] = None, space_attr: Optional[str] = None, default_space: Optional[str] = None, validate_name: bool = True
) -> None:
"""The fullname is determined in one of the following ways:
- A name and namespace are both specified. For example, one might use "name": "X",
"namespace": "org.foo" to indicate the fullname org.foo.X.
- A fullname is specified. If the name specified contains a dot,
then it is assumed to be a fullname, and any namespace also specified is ignored.
For example, use "name": "org.foo.X" to indicate the fullname org.foo.X.
- A name only is specified, i.e., a name that contains no dots.
In this case the namespace is taken from the most tightly enclosing schema or protocol.
For example, if "name": "X" is specified, and this occurs within a field of
the record definition of org.foo.Y, then the fullname is org.foo.X.
If there is no enclosing namespace then the null namespace is used.
References to previously defined names are as in the latter two cases above:
if they contain a dot they are a fullname,
if they do not contain a dot, the namespace is the namespace of the enclosing definition.
@arg name_attr: name value read in schema or None.
@arg space_attr: namespace value read in schema or None. The empty string may be used as a namespace
to indicate the null namespace.
@arg default_space: the current default space or None.
"""
if name_attr is None:
return
if name_attr == "":
raise avro.errors.SchemaParseException("Name must not be the empty string.")
# The empty string may be used as a namespace to indicate the null namespace.
self._name = name_attr.split(".")[-1]
self._full = (
name_attr
if "." in name_attr or space_attr == "" or not (space_attr or default_space)
else f"{space_attr or default_space!s}.{name_attr!s}"
)
if validate_name:
_validate_fullname(self._full)
def __eq__(self, other: object) -> bool:
"""Equality of names is defined on the fullname and is case-sensitive."""
return hasattr(other, "fullname") and self.fullname == getattr(other, "fullname")
@property
def name(self) -> Optional[str]:
return self._name
@property
def fullname(self) -> Optional[str]:
return self._full
@property
def space(self) -> Optional[str]:
"""Back out a namespace from full name."""
full = self._full or ""
return full.rsplit(".", 1)[0] if "." in full else None
def get_space(self) -> Optional[str]:
warnings.warn("Name.get_space() is deprecated in favor of Name.space")
return self.space
class Names:
"""Track name set and default namespace during parsing."""
names: Dict[str, "NamedSchema"]
def __init__(self, default_namespace: Optional[str] = None, validate_names: bool = True) -> None:
self.names = {}
self.default_namespace = default_namespace
self.validate_names = validate_names
def has_name(self, name_attr: str, space_attr: Optional[str] = None) -> bool:
test = Name(name_attr, space_attr, self.default_namespace, validate_name=self.validate_names).fullname
return test in self.names
def get_name(self, name_attr: str, space_attr: Optional[str] = None) -> Optional["NamedSchema"]:
test = Name(name_attr, space_attr, self.default_namespace, validate_name=self.validate_names).fullname
return None if test is None else self.names.get(test)
def prune_namespace(self, properties: Dict[str, object]) -> Dict[str, object]:
"""given a properties, return properties with namespace removed if
it matches the own default namespace"""
if self.default_namespace is None:
# I have no default -- no change
return properties
if "namespace" not in properties:
# he has no namespace - no change
return properties
if properties["namespace"] != self.default_namespace:
# we're different - leave his stuff alone
return properties
# we each have a namespace and it's redundant. delete his.
prunable = properties.copy()
del prunable["namespace"]
return prunable
def add_name(self, name_attr: str, space_attr: Optional[str], new_schema: "NamedSchema") -> Name:
"""
Add a new schema object to the name set.
@arg name_attr: name value read in schema
@arg space_attr: namespace value read in schema.
@return: the Name that was just added.
"""
to_add = Name(name_attr, space_attr, self.default_namespace, validate_name=self.validate_names)
if to_add.fullname in PRIMITIVE_TYPES:
raise avro.errors.SchemaParseException(f"{to_add.fullname} is a reserved type name.")
if to_add.fullname in self.names:
raise avro.errors.SchemaParseException(f'The name "{to_add.fullname}" is already in use.')
if to_add.fullname is None:
raise avro.errors.SchemaParseException(f'The name built from "{space_attr}.{name_attr}" is None')
self.names[to_add.fullname] = new_schema
return to_add
|
