talexander/infra_snippets
1
0
Fork 0
Code Issues Pull Requests Releases Activity

Add a schema for source location.

Browse Source
This commit is contained in:
Tom Alexander 2025-05-05 22:49:06 -04:00
parent c6f417f323
commit f91d6774ca
Signed by: talexander
GPG Key ID: D3A179C9A53C0EDE
1 changed files with 13 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
google_cloud_logging/log_stream/log_stream/log_entry.py
View File

@ -28,8 +28,8 @@ class BaseLogEntry(BaseModel):
span_id: Annotated[str | None, Field(alias="spanId", default=None)] span_id: Annotated[str | None, Field(alias="spanId", default=None)]
trace_sampled: Annotated[bool | None, Field(alias="traceSampled", default=None)] trace_sampled: Annotated[bool | None, Field(alias="traceSampled", default=None)]
source_location: Annotated[ source_location: Annotated[
dict | None, Field(alias="sourceLocation", default=None) SourceLocation | None, Field(alias="sourceLocation", default=None)
] # https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#LogEntrySourceLocation ]
split: Annotated[ split: Annotated[
dict | None, Field(default=None) dict | None, Field(default=None)
] # https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#LogSplit ] # https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#LogSplit
@ -56,6 +56,12 @@ class JsonPayloadLogEntry(BaseLogEntry):
json_payload: Annotated[dict, Field(alias="jsonPayload")] json_payload: Annotated[dict, Field(alias="jsonPayload")]
LogEntry = Annotated[
Union[ProtoPayloadLogEntry, JsonPayloadLogEntry, TextPayloadLogEntry, BaseLogEntry],
Field(),
]
class LogSeverity(str, Enum): class LogSeverity(str, Enum):
# https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#LogSeverity # https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#LogSeverity
DEFAULT = "DEFAULT" DEFAULT = "DEFAULT"
@ -69,10 +75,11 @@ class LogSeverity(str, Enum):
EMERGENCY = "EMERGENCY" EMERGENCY = "EMERGENCY"
LogEntry = Annotated[ class SourceLocation(BaseModel):
Union[ProtoPayloadLogEntry, JsonPayloadLogEntry, TextPayloadLogEntry, BaseLogEntry], # https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#LogEntrySourceLocation
Field(), file: str
] line: str
function: Annotated[str | None, Field(default=None)]
def create_log_entry(entry: dict) -> LogEntry: def create_log_entry(entry: dict) -> LogEntry: