Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/cloud/logging_v2/handlers/transports/sync.py: 53%

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. 

14 

15"""Transport for Python logging handler. 

16 

17Logs directly to the Cloud Logging API with a synchronous call. 

18""" 

19from google.cloud.logging_v2 import _helpers 

20from google.cloud.logging_v2.handlers.transports.base import Transport 

21from google.cloud.logging_v2.logger import _GLOBAL_RESOURCE 

22 

23 

24class SyncTransport(Transport): 

25 """Basic sychronous transport. 

26 

27 Uses this library's Logging client to directly make the API call. 

28 """ 

29 

30 def __init__(self, client, name, resource=_GLOBAL_RESOURCE, **kwargs): 

31 """ 

32 Args: 

33 client (~logging_v2.client.Client): 

34 The Logging client. 

35 name (str): The name of the lgoger. 

36 resource (Optional[Resource|dict]): The default monitored resource to associate 

37 with logs when not specified 

38 """ 

39 self.logger = client.logger(name, resource=resource) 

40 

41 def send(self, record, message, **kwargs): 

42 """Overrides transport.send(). 

43 

44 Args: 

45 record (logging.LogRecord): 

46 Python log record that the handler was called with. 

47 message (str or dict): The message from the ``LogRecord`` after being 

48 formatted by the associated log formatters. 

49 kwargs: Additional optional arguments for the logger 

50 """ 

51 # set python logger name as label if missing 

52 labels = kwargs.pop("labels", {}) 

53 if record.name: 

54 labels["python_logger"] = labels.get("python_logger", record.name) 

55 # send log synchronously 

56 self.logger.log( 

57 message, 

58 severity=_helpers._normalize_severity(record.levelno), 

59 labels=labels, 

60 **kwargs, 

61 ) 

62 

63 def close(self): 

64 """Closes the transport and cleans up resources used by it. 

65 

66 This call is usually followed up by cleaning up the reference to the transport. 

67 """ 

68 self.logger = None