Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/google/cloud/logging_v2/entries.py: 36%
141 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 07:30 +0000
1# Copyright 2016 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
15"""Log entries within the Google Cloud Logging API."""
17import collections
18import json
19import re
21from google.protobuf.any_pb2 import Any
22from google.protobuf.json_format import MessageToDict
23from google.protobuf.json_format import Parse
25from google.cloud.logging_v2.resource import Resource
26from google.cloud._helpers import _name_from_project_path
27from google.cloud._helpers import _rfc3339_nanos_to_datetime
28from google.cloud._helpers import _datetime_to_rfc3339
30# import officially supported proto definitions
31import google.cloud.audit.audit_log_pb2 # noqa: F401
32import google.cloud.appengine_logging # noqa: F401
33from google.iam.v1.logging import audit_data_pb2 # noqa: F401
35_GLOBAL_RESOURCE = Resource(type="global", labels={})
38_LOGGER_TEMPLATE = re.compile(
39 r"""
40 projects/ # static prefix
41 (?P<project>[^/]+) # initial letter, wordchars + hyphen
42 /logs/ # static midfix
43 (?P<name>[^/]+) # initial letter, wordchars + allowed punc
44""",
45 re.VERBOSE,
46)
49def logger_name_from_path(path, project=None):
50 """Validate a logger URI path and get the logger name.
52 Args:
53 path (str): URI path for a logger API request
54 project (str): The project the path is expected to belong to
56 Returns:
57 str: Logger name parsed from ``path``.
59 Raises:
60 ValueError: If the ``path`` is ill-formed of if the project
61 from ``path`` does not agree with the ``project`` passed in.
62 """
63 return _name_from_project_path(path, project, _LOGGER_TEMPLATE)
66def _int_or_none(value):
67 """Helper: return an integer or ``None``."""
68 if value is not None:
69 value = int(value)
70 return value
73_LOG_ENTRY_FIELDS = ( # (name, default)
74 ("log_name", None),
75 ("labels", None),
76 ("insert_id", None),
77 ("severity", None),
78 ("http_request", None),
79 ("timestamp", None),
80 ("resource", _GLOBAL_RESOURCE),
81 ("trace", None),
82 ("span_id", None),
83 ("trace_sampled", None),
84 ("source_location", None),
85 ("operation", None),
86 ("logger", None),
87 ("payload", None),
88)
91_LogEntryTuple = collections.namedtuple(
92 "LogEntry", (field for field, _ in _LOG_ENTRY_FIELDS)
93)
95_LogEntryTuple.__new__.__defaults__ = tuple(default for _, default in _LOG_ENTRY_FIELDS)
98_LOG_ENTRY_PARAM_DOCSTRING = """\
100 Args:
101 log_name (str): The name of the logger used to post the entry.
102 labels (Optional[dict]): Mapping of labels for the entry
103 insert_id (Optional[str]): The ID used to identify an entry
104 uniquely.
105 severity (Optional[str]): The severity of the event being logged.
106 http_request (Optional[dict]): Info about HTTP request associated
107 with the entry.
108 timestamp (Optional[datetime.datetime]): Timestamp for the entry.
109 resource (Optional[google.cloud.logging_v2.resource.Resource]):
110 Monitored resource of the entry.
111 trace (Optional[str]): Trace ID to apply to the entry.
112 span_id (Optional[str]): Span ID within the trace for the log
113 entry. Specify the trace parameter if ``span_id`` is set.
114 trace_sampled (Optional[bool]): The sampling decision of the trace
115 associated with the log entry.
116 source_location (Optional[dict]): Location in source code from which
117 the entry was emitted.
118 operation (Optional[dict]): Additional information about a potentially
119 long-running operation associated with the log entry.
120 logger (logging_v2.logger.Logger): the logger used
121 to write the entry.
122"""
124_LOG_ENTRY_SEE_ALSO_DOCSTRING = """\
126 See:
127 https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry
128"""
131class LogEntry(_LogEntryTuple):
132 __doc__ = (
133 """
134 Log entry.
136 """
137 + _LOG_ENTRY_PARAM_DOCSTRING
138 + _LOG_ENTRY_SEE_ALSO_DOCSTRING
139 )
141 received_timestamp = None
143 @classmethod
144 def _extract_payload(cls, resource):
145 """Helper for :meth:`from_api_repr`"""
146 return None
148 @classmethod
149 def from_api_repr(cls, resource, client, *, loggers=None):
150 """Construct an entry given its API representation
152 Args:
153 resource (dict): text entry resource representation returned from
154 the API
155 client (~logging_v2.client.Client):
156 Client which holds credentials and project configuration.
157 loggers (Optional[dict]):
158 A mapping of logger fullnames -> loggers. If not
159 passed, the entry will have a newly-created logger if possible,
160 or an empty logger field if not.
162 Returns:
163 google.cloud.logging.entries.LogEntry: Log entry parsed from ``resource``.
164 """
165 if loggers is None:
166 loggers = {}
167 logger_fullname = resource["logName"]
168 logger = loggers.get(logger_fullname)
169 if logger is None:
170 # attempt to create a logger if possible
171 try:
172 logger_name = logger_name_from_path(logger_fullname, client.project)
173 logger = loggers[logger_fullname] = client.logger(logger_name)
174 except ValueError:
175 # log name is not scoped to a project. Leave logger as None
176 pass
177 payload = cls._extract_payload(resource)
178 insert_id = resource.get("insertId")
179 timestamp = resource.get("timestamp")
180 if timestamp is not None:
181 timestamp = _rfc3339_nanos_to_datetime(timestamp)
182 labels = resource.get("labels")
183 severity = resource.get("severity")
184 http_request = resource.get("httpRequest")
185 trace = resource.get("trace")
186 span_id = resource.get("spanId")
187 trace_sampled = resource.get("traceSampled")
188 source_location = resource.get("sourceLocation")
189 if source_location is not None:
190 line = source_location.pop("line", None)
191 source_location["line"] = _int_or_none(line)
192 operation = resource.get("operation")
194 monitored_resource_dict = resource.get("resource")
195 monitored_resource = None
196 if monitored_resource_dict is not None:
197 monitored_resource = Resource._from_dict(monitored_resource_dict)
199 inst = cls(
200 log_name=logger_fullname,
201 insert_id=insert_id,
202 timestamp=timestamp,
203 labels=labels,
204 severity=severity,
205 http_request=http_request,
206 resource=monitored_resource,
207 trace=trace,
208 span_id=span_id,
209 trace_sampled=trace_sampled,
210 source_location=source_location,
211 operation=operation,
212 logger=logger,
213 payload=payload,
214 )
215 received = resource.get("receiveTimestamp")
216 if received is not None:
217 inst.received_timestamp = _rfc3339_nanos_to_datetime(received)
218 return inst
220 def to_api_repr(self):
221 """API repr (JSON format) for entry."""
222 info = {}
223 if self.log_name is not None:
224 info["logName"] = self.log_name
225 if self.resource is not None:
226 info["resource"] = self.resource._to_dict()
227 if self.labels is not None:
228 info["labels"] = self.labels
229 if self.insert_id is not None:
230 info["insertId"] = self.insert_id
231 if self.severity is not None:
232 info["severity"] = self.severity
233 if self.http_request is not None:
234 info["httpRequest"] = self.http_request
235 if self.timestamp is not None:
236 info["timestamp"] = _datetime_to_rfc3339(self.timestamp)
237 if self.trace is not None:
238 info["trace"] = self.trace
239 if self.span_id is not None:
240 info["spanId"] = self.span_id
241 if self.trace_sampled is not None:
242 info["traceSampled"] = self.trace_sampled
243 if self.source_location is not None:
244 source_location = self.source_location.copy()
245 source_location["line"] = str(source_location.pop("line", 0))
246 info["sourceLocation"] = source_location
247 if self.operation is not None:
248 info["operation"] = self.operation
249 return info
252class TextEntry(LogEntry):
253 __doc__ = (
254 """
255 Log entry with text payload.
257 """
258 + _LOG_ENTRY_PARAM_DOCSTRING
259 + """
261 payload (str): payload for the log entry.
262 """
263 + _LOG_ENTRY_SEE_ALSO_DOCSTRING
264 )
266 @classmethod
267 def _extract_payload(cls, resource):
268 """Helper for :meth:`from_api_repr`"""
269 return resource["textPayload"]
271 def to_api_repr(self):
272 """API repr (JSON format) for entry."""
273 info = super(TextEntry, self).to_api_repr()
274 info["textPayload"] = self.payload
275 return info
278class StructEntry(LogEntry):
279 __doc__ = (
280 """
281 Log entry with JSON payload.
283 """
284 + _LOG_ENTRY_PARAM_DOCSTRING
285 + """
287 payload (dict): payload for the log entry.
288 """
289 + _LOG_ENTRY_SEE_ALSO_DOCSTRING
290 )
292 @classmethod
293 def _extract_payload(cls, resource):
294 """Helper for :meth:`from_api_repr`"""
295 return resource["jsonPayload"]
297 def to_api_repr(self):
298 """API repr (JSON format) for entry."""
299 info = super(StructEntry, self).to_api_repr()
300 info["jsonPayload"] = self.payload
301 return info
304class ProtobufEntry(LogEntry):
305 __doc__ = (
306 """
307 Log entry with protobuf message payload.
309 """
310 + _LOG_ENTRY_PARAM_DOCSTRING
311 + """
313 payload (google.protobuf.Message): payload for the log entry.
314 """
315 + _LOG_ENTRY_SEE_ALSO_DOCSTRING
316 )
318 @classmethod
319 def _extract_payload(cls, resource):
320 """Helper for :meth:`from_api_repr`"""
321 return resource["protoPayload"]
323 @property
324 def payload_pb(self):
325 if isinstance(self.payload, Any):
326 return self.payload
328 @property
329 def payload_json(self):
330 if isinstance(self.payload, collections.abc.Mapping):
331 return self.payload
333 def to_api_repr(self):
334 """API repr (JSON format) for entry."""
335 info = super(ProtobufEntry, self).to_api_repr()
336 proto_payload = None
337 if self.payload_json:
338 proto_payload = dict(self.payload_json)
339 elif self.payload_pb:
340 proto_payload = MessageToDict(self.payload_pb)
341 info["protoPayload"] = proto_payload
342 return info
344 def parse_message(self, message):
345 """Parse payload into a protobuf message.
347 Mutates the passed-in ``message`` in place.
349 Args:
350 message (google.protobuf.Message): the message to be logged
351 """
352 # NOTE: This assumes that ``payload`` is already a deserialized
353 # ``Any`` field and ``message`` has come from an imported
354 # ``pb2`` module with the relevant protobuf message type.
355 Parse(json.dumps(self.payload), message)