Source code for botocraft.services.logs

# This file is automatically generated by botocraft.  Do not edit directly.
# mypy: disable-error-code="index, override, assignment, union-attr, misc"
from pydantic import Field
from botocraft.mixins.logs import convert_log_group_tags
from .abstract import (
    Boto3Model,
    ReadonlyBoto3Model,
    PrimaryBoto3Model,
    ReadonlyPrimaryBoto3Model,
    Boto3ModelManager,
    ReadonlyBoto3ModelManager,
)
from botocraft.mixins.logs import LogGroupManagerMixin
from botocraft.mixins.logs import convert_log_groups_tags
from typing import ClassVar, Literal, Any, cast
from .abstract import PrimaryBoto3ModelQuerySet
from botocraft.mixins.tags import TagsDictMixin
from datetime import datetime
import builtins
from botocraft.services.common import Tag
from botocraft.mixins.logs import create_log_group_extended

# ===============
# Managers
# ===============


[docs]class LogGroupManager(LogGroupManagerMixin, Boto3ModelManager): service_name: str = "logs"
[docs] @create_log_group_extended def create(self, model: "LogGroup", tags: "dict[str, str] | None" = None) -> None: """ Creates a log group with the specified name. You can create up to 1,000,000 log groups per Region per account. Args: model: The :py:class:`LogGroup` to create. Keyword Args: tags: The key-value pairs to use for the tags. """ data = model.model_dump(exclude_none=True, by_alias=True) args = dict( logGroupName=data.get("logGroupName"), kmsKeyId=data.get("kmsKeyId"), tags=self.serialize(tags), logGroupClass=data.get("logGroupClass"), deletionProtectionEnabled=data.get("deletionProtectionEnabled"), ) self.client.create_log_group(**{k: v for k, v in args.items() if v is not None})
[docs] def delete(self, logGroupName: str) -> None: """ Deletes the specified log group and permanently deletes all the archived log events associated with the log group. Args: logGroupName: The name of the log group. """ args: dict[str, Any] = dict(logGroupName=self.serialize(logGroupName)) self.client.delete_log_group(**{k: v for k, v in args.items() if v is not None})
[docs] @convert_log_group_tags def get( self, logGroupIdentifier: str, *, accountIdentifiers: "builtins.list[str] | None" = None, logGroupNamePattern: "str | None" = None, limit: "int | None" = None, includeLinkedAccounts: "bool | None" = None, ) -> "LogGroup | None": """ Returns information about log groups, including data sources that ingest into each log group. You can return all your log groups or filter the results by prefix. The results are ASCII-sorted by log group name. Args: logGroupIdentifier: The identifier of the log group to get. If this is in the same account you are querying, you can use the log group name, if you also set ````includeLinkedAccounts```` to ````true````, you must use the log group ARN. Keyword Args: accountIdentifiers: When ``includeLinkedAccounts`` is set to ``true``, use this parameter to specify the list of accounts to search. You can specify as many as 20 account IDs in the array. logGroupNamePattern: If you specify a string for this parameter, the operation returns only log groups that have names that match the string based on a case-sensitive substring search. For example, if you specify ``DataLogs``, log groups named ``DataLogs``, ``aws/DataLogs``, and ``GroupDataLogs`` would match, but ``datalogs``, ``Data/log/s`` and ``Groupdata`` would not match. limit: The maximum number of items returned. If you don't specify a value, the default is up to 50 items. includeLinkedAccounts: If you are using a monitoring account, set this to ``true`` to have the operation return log groups in the accounts listed in ``accountIdentifiers``. """ args: dict[str, Any] = dict( accountIdentifiers=self.serialize(accountIdentifiers), logGroupNamePattern=self.serialize(logGroupNamePattern), limit=self.serialize(limit), includeLinkedAccounts=self.serialize(includeLinkedAccounts), logGroupIdentifiers=self.serialize([logGroupIdentifier]), ) _response = self.client.describe_log_groups( **{k: v for k, v in args.items() if v is not None} ) response = DescribeLogGroupsResponse(**_response) if response and response.logGroups: self.sessionize(response.logGroups[0]) return response.logGroups[0] return None
[docs] @convert_log_groups_tags def list( self, *, accountIdentifiers: "builtins.list[str] | None" = None, logGroupNamePrefix: "str | None" = None, logGroupNamePattern: "str | None" = None, limit: "int | None" = None, includeLinkedAccounts: "bool | None" = None, logGroupClass: "Literal['STANDARD', 'INFREQUENT_ACCESS', 'DELIVERY'] | None" = None, logGroupIdentifiers: "builtins.list[str] | None" = None, ) -> PrimaryBoto3ModelQuerySet: """ Returns information about log groups, including data sources that ingest into each log group. You can return all your log groups or filter the results by prefix. The results are ASCII-sorted by log group name. Keyword Args: accountIdentifiers: When ``includeLinkedAccounts`` is set to ``true``, use this parameter to specify the list of accounts to search. You can specify as many as 20 account IDs in the array. logGroupNamePrefix: The prefix to match. logGroupNamePattern: If you specify a string for this parameter, the operation returns only log groups that have names that match the string based on a case-sensitive substring search. For example, if you specify ``DataLogs``, log groups named ``DataLogs``, ``aws/DataLogs``, and ``GroupDataLogs`` would match, but ``datalogs``, ``Data/log/s`` and ``Groupdata`` would not match. limit: The maximum number of items returned. If you don't specify a value, the default is up to 50 items. includeLinkedAccounts: If you are using a monitoring account, set this to ``true`` to have the operation return log groups in the accounts listed in ``accountIdentifiers``. logGroupClass: Use this parameter to limit the results to only those log groups in the specified log group class. If you omit this parameter, log groups of all classes can be returned. logGroupIdentifiers: Use this array to filter the list of log groups returned. If you specify this parameter, the only other filter that you can choose to specify is ``includeLinkedAccounts``. """ paginator = self.client.get_paginator("describe_log_groups") args: dict[str, Any] = dict( accountIdentifiers=self.serialize(accountIdentifiers), logGroupNamePrefix=self.serialize(logGroupNamePrefix), logGroupNamePattern=self.serialize(logGroupNamePattern), limit=self.serialize(limit), includeLinkedAccounts=self.serialize(includeLinkedAccounts), logGroupClass=self.serialize(logGroupClass), logGroupIdentifiers=self.serialize(logGroupIdentifiers), ) response_iterator = paginator.paginate( **{k: v for k, v in args.items() if v is not None} ) results = [] for _response in response_iterator: if list(_response.keys()) == ["ResponseMetadata"]: break if "ResponseMetadata" in _response: del _response["ResponseMetadata"] response = DescribeLogGroupsResponse(**_response) if response.logGroups: results.extend(response.logGroups) else: if getattr(response, "NextToken", None): continue break self.sessionize(results) if results and isinstance(results[0], Boto3Model): return PrimaryBoto3ModelQuerySet(results) return results
[docs] def get_fields( self, *, logGroupName: "str | None" = None, time: "int | None" = None, logGroupIdentifier: "str | None" = None, ) -> "builtins.list[LogGroupField]": """ Returns a list of the fields that are included in log events in the specified log group. Includes the percentage of log events that contain each field. The search is limited to a time period that you specify. Keyword Args: logGroupName: The name of the log group to search. time: The time to set as the center of the query. If you specify ``time``, the 8 minutes before and 8 minutes after this time are searched. If you omit ``time``, the most recent 15 minutes up to the current time are searched. logGroupIdentifier: Specify either the name or ARN of the log group to view. If the log group is in a source account and you are using a monitoring account, you must specify the ARN. """ args: dict[str, Any] = dict( logGroupName=self.serialize(logGroupName), time=self.serialize(time), logGroupIdentifier=self.serialize(logGroupIdentifier), ) _response = self.client.get_log_group_fields( **{k: v for k, v in args.items() if v is not None} ) response = GetLogGroupFieldsResponse(**_response) results: "builtins.list[LogGroupField]" = None if response is not None: results = response.logGroupFields self.sessionize(results) return cast("builtins.list[LogGroupField]", results)
# ============== # Service Models # ==============
[docs]class LogGroup(TagsDictMixin, PrimaryBoto3Model): """ Represents a log group. """ tag_class: ClassVar[type[Boto3Model]] = Tag manager_class: ClassVar[type[Boto3ModelManager]] = LogGroupManager Arn: str = Field(default=None, alias="arn") """ The Amazon Resource Name (ARN) of the log group. This version of the ARN includes a trailing ``:*`` after the log group name. """ logGroupName: "str | None" = None """ The name of the log group. """ creationTime: int = Field(default=None, frozen=True) """ The creation time of the log group, expressed as the number of milliseconds after Jan 1, 1970 00:00:00 UTC. """ retentionInDays: int = Field(default=None, frozen=True) """ The number of days to retain the log events in the specified log group. Possible values are: 1, 3, 5, 7, 14, 30, 60, 90, 120, 150, 180, 365, 400, 545, 731, 1096, 1827, 2192, 2557, 2922, 3288, and 3653. """ metricFilterCount: int = Field(default=None, frozen=True) """ The number of metric filters. """ storedBytes: int = Field(default=None, frozen=True) """ The number of bytes stored. """ kmsKeyId: "str | None" = None """ The Amazon Resource Name (ARN) of the KMS key to use when encrypting log data. """ dataProtectionStatus: Literal["ACTIVATED", "DELETED", "ARCHIVED", "DISABLED"] = ( Field(default=None, frozen=True) ) """ Displays whether this log group has a protection policy, or whether it had one in the past. For more information, see `P utDataProtectionPolicy <https://docs.aws.amazon.com/AmazonCloudWatchLogs/latest/APIReference/API_PutDataProtectionPolicy .html>`_. """ inheritedProperties: "builtins.list[Literal['ACCOUNT_DATA_PROTECTION']]" = Field( default_factory=list, frozen=True ) """ Displays all the properties that this log group has inherited from account-level settings. """ logGroupClass: "Literal['STANDARD', 'INFREQUENT_ACCESS', 'DELIVERY'] | None" = None """ This specifies the log group class for this log group. There are three classes: """ logGroupArn: str = Field(default=None, frozen=True) """ The Amazon Resource Name (ARN) of the log group. This version of the ARN doesn't include a trailing ``:*`` after the log group name. """ deletionProtectionEnabled: "bool | None" = None """ Indicates whether deletion protection is enabled for this log group. When enabled, deletion protection blocks all deletion operations until it is explicitly disabled. """ bearerTokenAuthenticationEnabled: bool = Field(default=None, frozen=True) """ Indicates whether bearer token authentication is enabled for this log group. When enabled, bearer token authentication is allowed on operations until it is explicitly disabled. """ Tags: "builtins.list[Tag] | None" = Field(default_factory=list) """ The tags associated with the log group. """ @property def pk(self) -> str | None: """ Return the primary key of the model. This is the value of the :py:attr:`Arn` attribute. Returns: The primary key of the model instance. """ return self.Arn @property def arn(self) -> str | None: """ Return the ARN of the model. This is the value of the :py:attr:`Arn` attribute. Returns: The ARN of the model instance. """ return self.Arn @property def name(self) -> str | None: """ Return the name of the model. This is the value of the :py:attr:`logGroupName` attribute. Returns: The name of the model instance. """ return self.logGroupName def __hash__(self) -> int: """ Return the hash of the model. This is the value of the :py:attr:`Arn` attribute. """ return hash(self.Arn)
# ======================= # Request/Response Models # =======================
[docs]class DescribeLogGroupsResponse(Boto3Model): logGroups: "builtins.list[LogGroup] | None" = Field(default_factory=list) """ An array of structures, where each structure contains the information about one log group. """ nextToken: "str | None" = None """ The token for the next set of items to return. The token expires after 24 hours. """
[docs]class LogGroupField(Boto3Model): """ The fields contained in log events found by a ``GetLogGroupFields`` operation, along with the percentage of queried log events in which each field appears. """ name: "str | None" = None """ The name of a log field. """ percent: "int | None" = None """ The percentage of log events queried that contained the field. """
[docs]class GetLogGroupFieldsResponse(Boto3Model): logGroupFields: "builtins.list[LogGroupField] | None" = Field(default_factory=list) """ The array of fields found in the query. Each object in the array contains the name of the field, along with the percentage of time it appeared in the log events that were queried. """