o `juU@sdZddlmZddlmZddlmZddlZddlm Z ddl m Z ddl mZmZmZmZddlmZe r>dd lmZeeZGd d d ZGd d d eeZGdddeeZGdddeeZGdddZGdddeeZGdddeeZdS)zwConfiguration management mixin for KafkaAdminClient. Also defines ConfigResource and ConfigResourceType data classes. ) annotations) defaultdict)IntEnumN) TYPE_CHECKING)AlterConfigsRequestDescribeConfigsRequestIncrementalAlterConfigsRequestListConfigResourcesRequest) EnumHelper)KafkaConnectionManagerc@s2eZdZUdZded<ded<ddZedd Zed d Zed d Z eddZ e dddZd?ddZed d!Zd@d#d$Zd@d%d&Zd'd(Zd)d*Zd+d,Ze dAd-d.Zed/d0ZdAd1d2ZdBd4d5ZdBd6d7ZdBd8d9ZdBd:d;Zd"S)CConfigAdminMixinzFMixin providing configuration management methods for KafkaAdminClient.r _managerdictconfigcCs,z |jjtWdStjyYdSw)NTF)r broker_version_data api_versionrErrorsIncompatibleBrokerVersionselfrR/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/admin/_configs.py(_check_incremental_alter_configs_support!s z9ConfigAdminMixin._check_incremental_alter_configs_supportcCs~|sgSt|ts tdg}|D])\}}t|tr!|\}}ntj|}}t|}|tjj kr4d}| |||fq|S)Nzcalter_configs requires configs as a dict of {key: (op, value)} or {key: value} (interpreted as SET)) isinstancer TypeErroritemstuple AlterConfigOpSET value_forDELETEvalueappendconfigsentriesnameop_valueopr!op_coderrr_incremental_configs_entries)s      z-ConfigAdminMixin._incremental_configs_entriescCst|tr t|S|SN)rrlistkeysr$rrr_describe_configs_entries<sz*ConfigAdminMixin._describe_configs_entriescCs|sgSt|tstdt|g}|D].\}}t|tr&|\}}ntj|}}t|}|tjj kr?t d|d| ||fq|S)Nz0configs should be a dict of {key: value}, found z?Non-incremental AlterConfigsRequest does not support operation z (SET only)) rrrtyperrrrrr! ValueErrorr"r#rrr_alter_configs_entries@s      z'ConfigAdminMixin._alter_configs_entriesc Csptt}g}|D]+}|jtjtjfvr.zt|j}Wn ty%tdw|| |q| |q||fS)NzHBroker resource names must be an integer or a string represented integer) rr, resource_typeConfigResourceTypeBROKER BROKER_LOGGERintr&r1r")config_resourcesbroker_resourcesother_resourcesconfig_resource broker_idrrr_group_config_resourcesRs  z(ConfigAdminMixin._group_config_resourcesFcs(|rdnd}tfdd|D||dS)Nrc"g|] }|j|j|jfqSr)r3r&r/r$.0crclsrr ez>ConfigAdminMixin._describe_configs_request..) resourcesinclude_synonyms min_version)r)rDr8rHrIrrCr_describe_configs_requestas  z*ConfigAdminMixin._describe_configs_requestcCsJd|vr t|dS|dr|tjurtjS|ddr tjSt|S)N config_source read_only is_defaultF)ConfigSourceType build_fromr4r5STATIC_BROKER_CONFIGgetDEFAULT_CONFIGdynamic_for_resource_type)rr3rrr_get_config_sourcejs  z#ConfigAdminMixin._get_config_sourcecCs4|tjkr |dr dS|||}||rdSdS)NrLTF)ConfigFilterTypeDYNAMICrT should_skip)rDr config_filterr3rKrrr_should_skip_configus   z$ConfigAdminMixin._should_skip_configcCsf|d}|||}|j|d<d|vr$|dD] }t|dj|d<qd|vr1t|dj|d<|S)Nr&rKsynonymssource config_type)poprTr&rN ConfigType)rDrr3r&rKsynonymrrr_process_config~s    z ConfigAdminMixin._process_configmodifiedc Cst|}tt}|D]4}|jD].}t|j}i}|jD]}|} | | |} | | ||s3| || <q|||j |j <qq t|Sr+)rUrOrrresultsr4r3r$to_dictr`rYr&lower resource_name) rD responsesrXretresponseresultr3resource_configs config_structrr&rrr#_describe_configs_process_responsess      z4ConfigAdminMixin._describe_configs_process_responsesc s||\}}g}|D]\}} || |} ||jj| |dIdHq|r;|||} ||j| IdH||||rLfdd|DSS)Nnode_idcs"g|] }|jj|jqSr)r3r&rdrAresourcergrrrErFz.)r=rrJr"r sendrl) rr8rHrXflatr9r:rfr<rGrequestrrqr_async_describe_configss    z(ConfigAdminMixin._async_describe_configscCs|j|j|||S)aFetch configuration parameters for one or more Kafka resources. Arguments: config_resources: An list of ConfigResource objects. Any keys in ConfigResource.configs dict will be used to filter the result. Setting the configs dict to None will get all values. An empty dict will get zero values (as per Kafka protocol). Keyword Arguments: include_synonyms (bool, optional): If True, return synonyms in response. Not supported by all versions. Default: False. config_filter (ConfigFilterType or str): Modified returns only keys that have non-default values; Dynamic returns all keys that can be modified with alter_configs; All returns all available keys. Default: Modified. Returns: dict of {resource_type (str): {resource_name (str): {config_key: {config data}}}} )r runru)rr8rHrXrrrdescribe_configss z!ConfigAdminMixin.describe_configscCs`t|j}|tjur|d|tt}|jD]}t|j }||j  |j qt|S)Nz5ListConfigResourcesRequest failed with response '{}'.)rfor_code error_codeNoErrorformatrr,r8r4r3r&rdr"rer)rh error_typergrpr3rrr'_list_config_resources_process_responses    z8ConfigAdminMixin._list_config_resources_process_responseNcsPg}|pgD] }t|}||jqt|d}|j|IdH}||S)N)resource_types)r4rOr"r!r r rrr})rr~ wire_typesrtrtrhrrr_async_list_config_resourcess    z-ConfigAdminMixin._async_list_config_resourcescCs|j|j|S)aList config resources known to the cluster. Useful for discovering resource types that have no separate enumeration API (e.g. ``CLIENT_METRICS``, ``GROUP``). For ``TOPIC`` and ``BROKER`` the data is also available via ``Metadata`` / cluster descriptions. Keyword Arguments: resource_types (list, optional): Filter by resource type. Each entry may be a :class:`ConfigResourceType` or its name (e.g. ``'TOPIC'``). If None or empty, the broker returns all supported types. Requires broker >= 4.1 for anything other than ``CLIENT_METRICS``. Returns: dict of {resource_type (str): [resource_name (str)]} )r rvr)rr~rrrlist_config_resourcessz&ConfigAdminMixin.list_config_resourcesc sdd|D}|j|dddIdH}g}t||D]$\}}i}|D]}||jvr8||d} | dur4q"| ||<q"||q|S)NcSg|] }t|j|jqSrConfigResourcer3r&rorrrrEzBConfigAdminMixin._get_missing_modified_configs..raTrXrsr!)ruzipr$r") rr8resource_lookupsdynamic_configsmissing_resource_configsrpdescribemissing config_key config_valuerrr_get_missing_modified_configss   z.ConfigAdminMixin._get_missing_modified_configscsR||IdH}t||D]\}}t|ts tdt||j|qdS)Nz&missing configs: expected dict, found )rrrrrr0r$update)rr8rrprrrr_add_missing_dynamic_configss z-ConfigAdminMixin._add_missing_dynamic_configscsfdd|D}|j|dddIdH}t||D]\}}t|jp!gt|}|r0td|qdS)NcSrrrrorrrrErz>ConfigAdminMixin._validate_dynamic_configs..dynamicTrzUnrecognized configs: )rursetr$r1)rr8rrrprunknownrrr_validate_dynamic_configssz*ConfigAdminMixin._validate_dynamic_configscs8|rtfdd|D|dStfdd|D|dS)Ncr?r)r3r&r*r$r@rCrrrE rFz;ConfigAdminMixin._alter_configs_request..)rG validate_onlycr?r)r3r&r2r$r@rCrrrErF)rr)rDr8r incrementalrrCr_alter_configs_requests  z'ConfigAdminMixin._alter_configs_requestcCs^tt}|D]$}|jdkrd}n tt|j|j}t|jj }||||j <qt|S)NrOK) rrrystrrrx error_messager4r3r&rdre)rfrgrhri result_typerrr _alter_configs_process_responsess z1ConfigAdminMixin._alter_configs_process_responsesc s||\}}g}|D]\}}||||} |jj| |dIdH} || jq|rC||||} |j| IdH} || j||S)Nrm)r=rrr rrextendrfr) rr8rrr9r:rfr<rGrtrhrrr_send_alter_configs_requestss  z-ConfigAdminMixin._send_alter_configs_requestsTcsN|r ||IdH|dur|}|s||IdH||||IdHSr+)rrrrrr8rraise_on_unknownrrrr_async_alter_configs,sz%ConfigAdminMixin._async_alter_configscC|j|j||||S)a6Alter configuration parameters of one or more Kafka resources. Arguments: config_resources: A list of ConfigResource objects. Each resource's ``configs`` must be a dict mapping config key to either ``(op, value)`` (where ``op`` is an :class:`AlterConfigOp`, its name, or its int value) or a bare value (interpreted as SET). For DELETE operations the value is ignored and sent as null. APPEND/SUBTRACT require broker >= 2.3. On older brokers only SET is supported; non-SET ops raise ValueError. On older brokers the client also fills in all other modified dynamic keys before submitting, since AlterConfigsRequest resets any omitted key to its default (be aware of the inherent race in that approach). validate_only (bool, optional): If True, changes are sent to broker for validation only. Changes will not be applied. Default: False raise_on_unknown (bool, optional): If True, raises ValueError if any config key is not recognized as a dynamic config for the resource. incremental (bool, optional): Set to True/False to force use of IncrementalAlterConfigs (True) or AlterConfigs (False). By Default, the admin client will use IncrementalAlterConfigs if supported by the broker, otherwise AlterConfigs. Returns: dict of {resource_type (str): {resource_name (str): Error/Result}} )r rvrrrrr alter_configs5szConfigAdminMixin.alter_configsc s|r ||IdH|dur|}|s2dd|D}||IdH}t||D]\}}||_q)n*dd|D} || IdH}t| |D]\}}||_qF|D] }dd|jD|_qP||||IdHS)NcSsg|]}|jr|qSrr.rorrrrEZz9ConfigAdminMixin._async_reset_configs..cSsg|]}|js|qSrr.rorrrrE`rcSsi|]}|tjdfqSr+)rr )rAkeyrrr fsz9ConfigAdminMixin._async_reset_configs..)rrrrr$r) rr8rrrpartial_resetsrrpr full_resetsrrr_async_reset_configsQs( z%ConfigAdminMixin._async_reset_configscCr)a,Reset configuration parameters of one or more Kafka resources to defaults. On 2.3+ brokers, the client will submit an IncrementalAlterConfigsRequest with op DELETE for each resource/key. On older brokers, the client will use submit an AlterConfigsRequest and attempt to include all modified dynamic config values for each resource except the keys marked for reset. (AlterConfigsRequest will reset any missing config key to its default). Arguments: config_resources: A list of ConfigResource objects. Each resource's ``configs`` should be a list or dict of config keys to reset. (if dict, the values are ignored). Returns: dict of {resource_type (str): {resource_name (str): Error/Result}} )r rvrrrrr reset_configsjszConfigAdminMixin.reset_configs)F)ra)FraF)Frar+)FF)FTN) __name__ __module__ __qualname____doc____annotations__r staticmethodr*r/r2r= classmethodrJrTrYr`rlrurwr}rrrrrrrrrrrrrrrrr sR                 r c@seZdZdZdZdZdZdS)rrr>N)rrrrr APPENDSUBTRACTrrrrr~s rc@s(eZdZdZdZdZdZdZddZdS) rUrr>rrcCs>|tjur | S|tjur|S|tjur|tjuSdS)NF)rUMODIFIED is_modifiedDEFAULTSTATICrNrP)rrKrrrrWs     zConfigFilterType.should_skipN) rrrALLrVrrrrWrrrrrUs rUc@s$eZdZdZdZdZdZdZdZdS)r4rrr N) rrrUNKNOWNTOPICr5r6CLIENT_METRICSGROUPrrrrr4sr4c@s*eZdZdZd ddZddZddZdS) ra A class for specifying config resources. Arguments: resource_type (ConfigResourceType): the type of kafka resource name (string): The name of the kafka resource configs ([key] or {key : value}): config keys (values required to alter) NcCs0t|ts tt|}||_||_||_dSr+)rr4rupperr3r&r$)rr3r&r$rrr__init__s  zConfigResource.__init__cCsd|jd|jS)NzConfigResource =)r&r3rrrr__str__szConfigResource.__str__cCsd|jd|jd|jdS)NzConfigResource(z, ))r3r&r$rrrr__repr__szConfigResource.__repr__r+)rrrrrrrrrrrrs   rc@s4eZdZdZdZdZdZdZdZdZ dZ d Z d Z d S) r^rr>rrrr N) rrrrBOOLEANSTRINGINTSHORTLONGDOUBLELISTCLASSPASSWORDrrrrr^sr^c@sDeZdZdZdZdZdZdZdZdZ dZ d Z d d Z e d d ZdS)rNrr>rrrrrrrcCs |jdvS)N)rrr)r!rrrrrs zConfigSourceType.is_modifiedcCsn|tjurtjS|tjurtjS|tjurtjS|tjur tjS|tj ur(tj S|tj ur0tj St d|)NzUnrecognized resource type )r4rrNrDYNAMIC_TOPIC_CONFIGr5DYNAMIC_BROKER_CONFIGr6DYNAMIC_BROKER_LOGGER_CONFIGrDYNAMIC_CLIENT_METRICS_CONFIGrDYNAMIC_GROUP_CONFIG RuntimeError)rDr3rrrrSs      z*ConfigSourceType.dynamic_for_resource_typeN)rrrrrrDYNAMIC_DEFAULT_BROKER_CONFIGrPrRrrrrrrSrrrrrNsrN) r __future__r collectionsrenumrloggingtypingr kafka.errorserrorsrkafka.protocol.adminrrrr kafka.utilr kafka.net.managerr getLoggerrlogr rrUr4rr^rNrrrrs*        d