
from typing import Optional

import torch
import torch.utils.checkpoint
from torch import Tensor, nn
from torch.nn import BCEWithLogitsLoss, CrossEntropyLoss, MSELoss

from ...activations import ACTaFN
from ...modeling_outputs import (
    BackboneOutput,
    BaseModelOutputWithNoAttention,
    BaseModelOutputWithPoolingAndNoAttention,
    ImageClassifierOutputWithNoAttention,
)
from ...modeling_utils import PreTrainedModel
from ...utils import (
    add_code_sample_docstrings,
    add_start_docstrings,
    add_start_docstrings_to_model_forward,
    logging,
    replace_return_docstrings,
)
from ...utils.backbone_utils import BackboneMixin
from .configuration_resnet import ResNetConfig


A__							: Optional[Any]     =       logging.get_logger(__name__)

# General docstring
A__							: Union[str, Any]     =       '''ResNetConfig'''

# Base docstring
A__							: Tuple     =       '''microsoft/resnet-50'''
A__							: List[str]     =       [1, 2_0_4_8, 7, 7]

# Image classification docstring
A__							: Optional[Any]     =       '''microsoft/resnet-50'''
A__							: Tuple     =       '''tiger cat'''

A__							: Dict     =       [
    '''microsoft/resnet-50''',
    # See all resnet models at https://huggingface.co/models?filter=resnet
]





class     lowercase							(	nn.Module       ):
					def __init__(		self			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__ = 3			,						SCREAMING_SNAKE_CASE__ = 1			,						SCREAMING_SNAKE_CASE__ = "relu"     ):
								"""simple docstring"""







								super().__init__()
								lowerCAmelCase__			:     Tuple					   =      nn.Convad(
								    __lowerCAmelCase			,						__lowerCAmelCase			,						kernel_size=__lowerCAmelCase			,						stride=__lowerCAmelCase			,						padding=kernel_size // 2			,						bias=__lowerCAmelCase     )
								lowerCAmelCase__			:     Dict					   =      nn.BatchNormad(__lowerCAmelCase     )
								lowerCAmelCase__			:     Optional[int]					   =      ACTaFN[activation] if activation is not None else nn.Identity()
					def lowercase_      (		self			,						SCREAMING_SNAKE_CASE__     ):
								"""simple docstring"""







								lowerCAmelCase__			:     Optional[int]					   =      self.convolution(__lowerCAmelCase     )
								lowerCAmelCase__			:     Optional[Any]					   =      self.normalization(__lowerCAmelCase     )
								lowerCAmelCase__			:     Dict					   =      self.activation(__lowerCAmelCase     )
								return hidden_state





class     lowercase							(	nn.Module       ):
					def __init__(		self			,						SCREAMING_SNAKE_CASE__     ):
								"""simple docstring"""







								super().__init__()
								lowerCAmelCase__			:     int					   =      ResNetConvLayer(
								    config.num_channels			,						config.embedding_size			,						kernel_size=7			,						stride=2			,						activation=config.hidden_act     )
								lowerCAmelCase__			:     Tuple					   =      nn.MaxPoolad(kernel_size=3			,						stride=2			,						padding=1     )
								lowerCAmelCase__			:     Tuple					   =      config.num_channels
					def lowercase_      (		self			,						SCREAMING_SNAKE_CASE__     ):
								"""simple docstring"""







								lowerCAmelCase__			:     Optional[int]					   =      pixel_values.shape[1]
								if num_channels != self.num_channels:
											raise ValueError(
											    '''Make sure that the channel dimension of the pixel values match with the one set in the configuration.'''     )
								lowerCAmelCase__			:     Tuple					   =      self.embedder(__lowerCAmelCase     )
								lowerCAmelCase__			:     str					   =      self.pooler(__lowerCAmelCase     )
								return embedding





class     lowercase							(	nn.Module       ):
					def __init__(		self			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__ = 2     ):
								"""simple docstring"""







								super().__init__()
								lowerCAmelCase__			:     str					   =      nn.Convad(__lowerCAmelCase			,						__lowerCAmelCase			,						kernel_size=1			,						stride=__lowerCAmelCase			,						bias=__lowerCAmelCase     )
								lowerCAmelCase__			:     List[Any]					   =      nn.BatchNormad(__lowerCAmelCase     )
					def lowercase_      (		self			,						SCREAMING_SNAKE_CASE__     ):
								"""simple docstring"""







								lowerCAmelCase__			:     Union[str, Any]					   =      self.convolution(__lowerCAmelCase     )
								lowerCAmelCase__			:     Dict					   =      self.normalization(__lowerCAmelCase     )
								return hidden_state





class     lowercase							(	nn.Module       ):
					def __init__(		self			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__ = 1			,						SCREAMING_SNAKE_CASE__ = "relu"     ):
								"""simple docstring"""







								super().__init__()
								lowerCAmelCase__			:     Dict					   =      in_channels != out_channels or stride != 1
								lowerCAmelCase__			:     List[str]					   =      (
								    ResNetShortCut(__lowerCAmelCase			,						__lowerCAmelCase			,						stride=__lowerCAmelCase     ) if should_apply_shortcut else nn.Identity()
								)
								lowerCAmelCase__			:     Dict					   =      nn.Sequential(
								    ResNetConvLayer(__lowerCAmelCase			,						__lowerCAmelCase			,						stride=__lowerCAmelCase     )			,						ResNetConvLayer(__lowerCAmelCase			,						__lowerCAmelCase			,						activation=__lowerCAmelCase     )			,						)
								lowerCAmelCase__			:     Union[str, Any]					   =      ACTaFN[activation]
					def lowercase_      (		self			,						SCREAMING_SNAKE_CASE__     ):
								"""simple docstring"""







								lowerCAmelCase__			:     List[str]					   =      hidden_state
								lowerCAmelCase__			:     List[Any]					   =      self.layer(__lowerCAmelCase     )
								lowerCAmelCase__			:     int					   =      self.shortcut(__lowerCAmelCase     )
								hidden_state += residual
								lowerCAmelCase__			:     Optional[Any]					   =      self.activation(__lowerCAmelCase     )
								return hidden_state





class     lowercase							(	nn.Module       ):
					def __init__(		self			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__ = 1			,						SCREAMING_SNAKE_CASE__ = "relu"			,						SCREAMING_SNAKE_CASE__ = 4     ):
								"""simple docstring"""







								super().__init__()
								lowerCAmelCase__			:     Optional[int]					   =      in_channels != out_channels or stride != 1
								lowerCAmelCase__			:     List[str]					   =      out_channels // reduction
								lowerCAmelCase__			:     List[str]					   =      (
								    ResNetShortCut(__lowerCAmelCase			,						__lowerCAmelCase			,						stride=__lowerCAmelCase     ) if should_apply_shortcut else nn.Identity()
								)
								lowerCAmelCase__			:     Optional[Any]					   =      nn.Sequential(
								    ResNetConvLayer(__lowerCAmelCase			,						__lowerCAmelCase			,						kernel_size=1     )			,						ResNetConvLayer(__lowerCAmelCase			,						__lowerCAmelCase			,						stride=__lowerCAmelCase     )			,						ResNetConvLayer(__lowerCAmelCase			,						__lowerCAmelCase			,						kernel_size=1			,						activation=__lowerCAmelCase     )			,						)
								lowerCAmelCase__			:     List[Any]					   =      ACTaFN[activation]
					def lowercase_      (		self			,						SCREAMING_SNAKE_CASE__     ):
								"""simple docstring"""







								lowerCAmelCase__			:     int					   =      hidden_state
								lowerCAmelCase__			:     int					   =      self.layer(__lowerCAmelCase     )
								lowerCAmelCase__			:     str					   =      self.shortcut(__lowerCAmelCase     )
								hidden_state += residual
								lowerCAmelCase__			:     List[str]					   =      self.activation(__lowerCAmelCase     )
								return hidden_state





class     lowercase							(	nn.Module       ):
					def __init__(		self			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__ = 2			,						SCREAMING_SNAKE_CASE__ = 2			,						):
								"""simple docstring"""







								super().__init__()

								lowerCAmelCase__			:     Tuple					   =      ResNetBottleNeckLayer if config.layer_type == '''bottleneck''' else ResNetBasicLayer

								lowerCAmelCase__			:     Any					   =      nn.Sequential(
								    # downsampling is done in the first layer with stride of 2
								    layer(__lowerCAmelCase			,						__lowerCAmelCase			,						stride=__lowerCAmelCase			,						activation=config.hidden_act     )			,						*[layer(__lowerCAmelCase			,						__lowerCAmelCase			,						activation=config.hidden_act     ) for _ in range(depth - 1     )]			,						)
					def lowercase_      (		self			,						SCREAMING_SNAKE_CASE__     ):
								"""simple docstring"""







								lowerCAmelCase__			:     Dict					   =      input
								for layer in self.layers:
											lowerCAmelCase__			:     Any					   =      layer(__lowerCAmelCase     )
								return hidden_state





class     lowercase							(	nn.Module       ):
					def __init__(		self			,						SCREAMING_SNAKE_CASE__     ):
								"""simple docstring"""







								super().__init__()
								lowerCAmelCase__			:     List[str]					   =      nn.ModuleList([]     )
								# based on `downsample_in_first_stage` the first layer of the first stage may or may not downsample the input
								self.stages.append(
								    ResNetStage(
								        __lowerCAmelCase			,						config.embedding_size			,						config.hidden_sizes[0]			,						stride=2 if config.downsample_in_first_stage else 1			,						depth=config.depths[0]			,						)     )
								lowerCAmelCase__			:     List[str]					   =      zip(config.hidden_sizes			,						config.hidden_sizes[1:]     )
								for (in_channels, out_channels), depth in zip(__lowerCAmelCase			,						config.depths[1:]     ):
											self.stages.append(ResNetStage(__lowerCAmelCase			,						__lowerCAmelCase			,						__lowerCAmelCase			,						depth=__lowerCAmelCase     )     )
					def lowercase_      (		self			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__ = False			,						SCREAMING_SNAKE_CASE__ = True     ):
								"""simple docstring"""







								lowerCAmelCase__			:     Dict					   =      () if output_hidden_states else None

								for stage_module in self.stages:
											if output_hidden_states:
														lowerCAmelCase__			:     int					   =      hidden_states + (hidden_state,)

											lowerCAmelCase__			:     Dict					   =      stage_module(__lowerCAmelCase     )

								if output_hidden_states:
											lowerCAmelCase__			:     str					   =      hidden_states + (hidden_state,)

								if not return_dict:
											return tuple(v for v in [hidden_state, hidden_states] if v is not None     )

								return BaseModelOutputWithNoAttention(
								    last_hidden_state=__lowerCAmelCase			,						hidden_states=__lowerCAmelCase			,						)





class     lowercase							(	__UpperCamelCase       ):
					__a											=     ResNetConfig
					__a											=     '''resnet'''
					__a											=     '''pixel_values'''
					__a											=     True
					def lowercase_      (		self			,						SCREAMING_SNAKE_CASE__     ):
								"""simple docstring"""







								if isinstance(__lowerCAmelCase			,						nn.Convad     ):
											nn.init.kaiming_normal_(module.weight			,						mode='''fan_out'''			,						nonlinearity='''relu'''     )
								elif isinstance(__lowerCAmelCase			,						(nn.BatchNormad, nn.GroupNorm)     ):
											nn.init.constant_(module.weight			,						1     )
											nn.init.constant_(module.bias			,						0     )
					def lowercase_      (		self			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__=False     ):
								"""simple docstring"""







								if isinstance(__lowerCAmelCase			,						__lowerCAmelCase     ):
											lowerCAmelCase__			:     int					   =      value





A__							: Dict     =       R'''
    This model is a PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) subclass. Use it
    as a regular PyTorch Module and refer to the PyTorch documentation for all matter related to general usage and
    behavior.

    Parameters:
        config ([`ResNetConfig`]): Model configuration class with all the parameters of the model.
            Initializing with a config file does not load the weights associated with the model, only the
            configuration. Check out the [`~PreTrainedModel.from_pretrained`] method to load the model weights.
'''

A__							: str     =       R'''
    Args:
        pixel_values (`torch.FloatTensor` of shape `(batch_size, num_channels, height, width)`):
            Pixel values. Pixel values can be obtained using [`AutoImageProcessor`]. See
            [`ConvNextImageProcessor.__call__`] for details.

        output_hidden_states (`bool`, *optional*):
            Whether or not to return the hidden states of all layers. See `hidden_states` under returned tensors for
            more detail.
        return_dict (`bool`, *optional*):
            Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple.
'''





@add_start_docstrings(
    """The bare ResNet model outputting raw features without any specific head on top."""						,		__UpperCamelCase						,		)
class     lowercase							(	__UpperCamelCase       ):
					def __init__(		self			,						SCREAMING_SNAKE_CASE__     ):
								"""simple docstring"""







								super().__init__(__lowerCAmelCase     )
								lowerCAmelCase__			:     Union[str, Any]					   =      config
								lowerCAmelCase__			:     List[str]					   =      ResNetEmbeddings(__lowerCAmelCase     )
								lowerCAmelCase__			:     Optional[int]					   =      ResNetEncoder(__lowerCAmelCase     )
								lowerCAmelCase__			:     Union[str, Any]					   =      nn.AdaptiveAvgPoolad((1, 1)     )
								# Initialize weights and apply final processing
								self.post_init()
					@add_start_docstrings_to_model_forward(__lowerCAmelCase     )
					@add_code_sample_docstrings(
					    checkpoint=_CHECKPOINT_FOR_DOC			,						output_type=__lowerCAmelCase			,						config_class=_CONFIG_FOR_DOC			,						modality='''vision'''			,						expected_output=_EXPECTED_OUTPUT_SHAPE			,						)
					def lowercase_      (		self			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__ = None			,						SCREAMING_SNAKE_CASE__ = None     ):
								"""simple docstring"""







								lowerCAmelCase__			:     Optional[Any]					   =      (
								    output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
								)
								lowerCAmelCase__			:     Any					   =      return_dict if return_dict is not None else self.config.use_return_dict

								lowerCAmelCase__			:     Any					   =      self.embedder(__lowerCAmelCase     )

								lowerCAmelCase__			:     Tuple					   =      self.encoder(
								    __lowerCAmelCase			,						output_hidden_states=__lowerCAmelCase			,						return_dict=__lowerCAmelCase     )

								lowerCAmelCase__			:     Dict					   =      encoder_outputs[0]

								lowerCAmelCase__			:     int					   =      self.pooler(__lowerCAmelCase     )

								if not return_dict:
											return (last_hidden_state, pooled_output) + encoder_outputs[1:]

								return BaseModelOutputWithPoolingAndNoAttention(
								    last_hidden_state=__lowerCAmelCase			,						pooler_output=__lowerCAmelCase			,						hidden_states=encoder_outputs.hidden_states			,						)





@add_start_docstrings(
    """
    ResNet Model with an image classification head on top (a linear layer on top of the pooled features), e.g. for
    ImageNet.
    """						,		__UpperCamelCase						,		)
class     lowercase							(	__UpperCamelCase       ):
					def __init__(		self			,						SCREAMING_SNAKE_CASE__     ):
								"""simple docstring"""







								super().__init__(__lowerCAmelCase     )
								lowerCAmelCase__			:     Union[str, Any]					   =      config.num_labels
								lowerCAmelCase__			:     Optional[int]					   =      ResNetModel(__lowerCAmelCase     )
								# classification head
								lowerCAmelCase__			:     int					   =      nn.Sequential(
								    nn.Flatten()			,						nn.Linear(config.hidden_sizes[-1]			,						config.num_labels     ) if config.num_labels > 0 else nn.Identity()			,						)
								# initialize weights and apply final processing
								self.post_init()
					@add_start_docstrings_to_model_forward(__lowerCAmelCase     )
					@add_code_sample_docstrings(
					    checkpoint=_IMAGE_CLASS_CHECKPOINT			,						output_type=__lowerCAmelCase			,						config_class=_CONFIG_FOR_DOC			,						expected_output=_IMAGE_CLASS_EXPECTED_OUTPUT			,						)
					def lowercase_      (		self			,						SCREAMING_SNAKE_CASE__ = None			,						SCREAMING_SNAKE_CASE__ = None			,						SCREAMING_SNAKE_CASE__ = None			,						SCREAMING_SNAKE_CASE__ = None			,						):
								"""simple docstring"""







								lowerCAmelCase__			:     Any					   =      return_dict if return_dict is not None else self.config.use_return_dict

								lowerCAmelCase__			:     Union[str, Any]					   =      self.resnet(__lowerCAmelCase			,						output_hidden_states=__lowerCAmelCase			,						return_dict=__lowerCAmelCase     )

								lowerCAmelCase__			:     int					   =      outputs.pooler_output if return_dict else outputs[1]

								lowerCAmelCase__			:     Any					   =      self.classifier(__lowerCAmelCase     )

								lowerCAmelCase__			:     Tuple					   =      None

								if labels is not None:
											if self.config.problem_type is None:
														if self.num_labels == 1:
																	lowerCAmelCase__			:     Union[str, Any]					   =      '''regression'''
														elif self.num_labels > 1 and (labels.dtype == torch.long or labels.dtype == torch.int):
																	lowerCAmelCase__			:     Tuple					   =      '''single_label_classification'''
														else:
																	lowerCAmelCase__			:     int					   =      '''multi_label_classification'''
											if self.config.problem_type == "regression":
														lowerCAmelCase__			:     Optional[int]					   =      MSELoss()
														if self.num_labels == 1:
																	lowerCAmelCase__			:     Optional[int]					   =      loss_fct(logits.squeeze()			,						labels.squeeze()     )
														else:
																	lowerCAmelCase__			:     int					   =      loss_fct(__lowerCAmelCase			,						__lowerCAmelCase     )
											elif self.config.problem_type == "single_label_classification":
														lowerCAmelCase__			:     Optional[Any]					   =      CrossEntropyLoss()
														lowerCAmelCase__			:     List[Any]					   =      loss_fct(logits.view(-1			,						self.num_labels     )			,						labels.view(-1     )     )
											elif self.config.problem_type == "multi_label_classification":
														lowerCAmelCase__			:     int					   =      BCEWithLogitsLoss()
														lowerCAmelCase__			:     List[str]					   =      loss_fct(__lowerCAmelCase			,						__lowerCAmelCase     )

								if not return_dict:
											lowerCAmelCase__			:     Union[str, Any]					   =      (logits,) + outputs[2:]
											return (loss,) + output if loss is not None else output

								return ImageClassifierOutputWithNoAttention(loss=__lowerCAmelCase			,						logits=__lowerCAmelCase			,						hidden_states=outputs.hidden_states     )





@add_start_docstrings(
    """
    ResNet backbone, to be used with frameworks like DETR and MaskFormer.
    """						,		__UpperCamelCase						,		)
class     lowercase							(	__UpperCamelCase						,		__UpperCamelCase       ):
					def __init__(		self			,						SCREAMING_SNAKE_CASE__     ):
								"""simple docstring"""







								super().__init__(__lowerCAmelCase     )
								super()._init_backbone(__lowerCAmelCase     )

								lowerCAmelCase__			:     Dict					   =      [config.embedding_size] + config.hidden_sizes
								lowerCAmelCase__			:     Tuple					   =      ResNetEmbeddings(__lowerCAmelCase     )
								lowerCAmelCase__			:     Dict					   =      ResNetEncoder(__lowerCAmelCase     )

								# initialize weights and apply final processing
								self.post_init()
					@add_start_docstrings_to_model_forward(__lowerCAmelCase     )
					@replace_return_docstrings(output_type=__lowerCAmelCase			,						config_class=_CONFIG_FOR_DOC     )
					def lowercase_      (		self			,						SCREAMING_SNAKE_CASE__			,						SCREAMING_SNAKE_CASE__ = None			,						SCREAMING_SNAKE_CASE__ = None     ):
								"""simple docstring"""







								lowerCAmelCase__			:     List[str]					   =      return_dict if return_dict is not None else self.config.use_return_dict
								lowerCAmelCase__			:     str					   =      (
								    output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
								)

								lowerCAmelCase__			:     List[Any]					   =      self.embedder(__lowerCAmelCase     )

								lowerCAmelCase__			:     Union[str, Any]					   =      self.encoder(__lowerCAmelCase			,						output_hidden_states=__lowerCAmelCase			,						return_dict=__lowerCAmelCase     )

								lowerCAmelCase__			:     Any					   =      outputs.hidden_states

								lowerCAmelCase__			:     Tuple					   =      ()
								for idx, stage in enumerate(self.stage_names     ):
											if stage in self.out_features:
														feature_maps += (hidden_states[idx],)

								if not return_dict:
											lowerCAmelCase__			:     Tuple					   =      (feature_maps,)
											if output_hidden_states:
														output += (outputs.hidden_states,)
											return output

								return BackboneOutput(
								    feature_maps=__lowerCAmelCase			,						hidden_states=outputs.hidden_states if output_hidden_states else None			,						attentions=__lowerCAmelCase			,						)










import multiprocessing
from typing import TYPE_CHECKING, Optional, Union

from .. import Dataset, Features, config
from ..formatting import query_table
from ..packaged_modules.sql.sql import Sql
from ..utils import logging
from .abc import AbstractDatasetInputStream


if TYPE_CHECKING:
					import sqlitea

					import sqlalchemy


class     A     (SCREAMING_SNAKE_CASE       ):
			'''simple docstring'''




			def __init__(     self  :       Any				,      __lowerCAmelCase  :       Union[str, "sqlalchemy.sql.Selectable"]				,      __lowerCAmelCase  :       Union[str, "sqlalchemy.engine.Connection", "sqlalchemy.engine.Engine", "sqlite3.Connection"]				,      __lowerCAmelCase  :       Optional[Features] = None				,      __lowerCAmelCase  :       str = None				,      __lowerCAmelCase  :       bool = False				,      **__lowerCAmelCase  :       Dict				,      )      ->      Any:







					"""simple docstring"""






					super().__init__(features=__lowerCAmelCase				,      cache_dir=__lowerCAmelCase				,      keep_in_memory=__lowerCAmelCase				,      **__lowerCAmelCase  )
					A__     =  Sql(
					    cache_dir=__lowerCAmelCase				,      features=__lowerCAmelCase				,      sql=__lowerCAmelCase				,      con=__lowerCAmelCase				,      **__lowerCAmelCase				,      )







			def   a_			(     self  :       Optional[int]  )      ->      Optional[int]:







					"""simple docstring"""






					A__     =  None
					A__     =  None
					A__     =  None
					A__     =  None

					self.builder.download_and_prepare(
					    download_config=__lowerCAmelCase				,      download_mode=__lowerCAmelCase				,      verification_mode=__lowerCAmelCase				,      base_path=__lowerCAmelCase				,      )

					# Build dataset for splits
					A__     =  self.builder.as_dataset(
					    split="""train"""				,      verification_mode=__lowerCAmelCase				,      in_memory=self.keep_in_memory  )
					return dataset


class     A     :
			'''simple docstring'''




			def __init__(     self  :       Optional[int]				,      __lowerCAmelCase  :       Dataset				,      __lowerCAmelCase  :       str				,      __lowerCAmelCase  :       Union[str, "sqlalchemy.engine.Connection", "sqlalchemy.engine.Engine", "sqlite3.Connection"]				,      __lowerCAmelCase  :       Optional[int] = None				,      __lowerCAmelCase  :       Optional[int] = None				,      **__lowerCAmelCase  :       List[Any]				,      )      ->      Any:







					"""simple docstring"""






					if num_proc is not None and num_proc <= 0:
							raise ValueError(f'num_proc {num_proc} must be an integer > 0.'  )

					A__     =  dataset
					A__     =  name
					A__     =  con
					A__     =  batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
					A__     =  num_proc
					A__     =  to_sql_kwargs






			def   a_			(     self  :       Any  )      ->      int:







					"""simple docstring"""






					A__     =  self.to_sql_kwargs.pop("""sql"""				,      __lowerCAmelCase  )
					A__     =  self.to_sql_kwargs.pop("""con"""				,      __lowerCAmelCase  )
					A__     =  self.to_sql_kwargs.pop("""index"""				,      __lowerCAmelCase  )

					A__     =  self._write(index=__lowerCAmelCase				,      **self.to_sql_kwargs  )
					return written






			def   a_			(     self  :       Union[str, Any]				,      __lowerCAmelCase  :       Optional[Any]  )      ->      Optional[Any]:







					"""simple docstring"""






					A__ ,     A__ ,     A__     =  args
					A__     =  {**to_sql_kwargs, """if_exists""": """append"""} if offset > 0 else to_sql_kwargs
					A__     =  query_table(
					    table=self.dataset.data				,      key=slice(__lowerCAmelCase				,      offset + self.batch_size  )				,      indices=self.dataset._indices				,      )
					A__     =  batch.to_pandas()
					A__     =  df.to_sql(self.name				,      self.con				,      index=__lowerCAmelCase				,      **__lowerCAmelCase  )
					return num_rows or len(__lowerCAmelCase  )







			def   a_			(     self  :       Optional[int]				,      __lowerCAmelCase  :       Tuple				,      **__lowerCAmelCase  :       Dict  )      ->      int:







					"""simple docstring"""






					A__     =  0

					if self.num_proc is None or self.num_proc == 1:
							for offset in logging.tqdm(
							    range(0				,      len(self.dataset  )				,      self.batch_size  )				,      unit="""ba"""				,      disable=not logging.is_progress_bar_enabled()				,      desc="""Creating SQL from Arrow format"""				,      ):
									written += self._batch_sql((offset, index, to_sql_kwargs)  )
					else:
							A__ ,     A__     =  len(self.dataset  ), self.batch_size
							with multiprocessing.Pool(self.num_proc  ) as pool:
									for num_rows in logging.tqdm(
									    pool.imap(
									        self._batch_sql				,      [(offset, index, to_sql_kwargs) for offset in range(0				,      __lowerCAmelCase				,      __lowerCAmelCase  )]				,      )				,      total=(num_rows // batch_size) + 1 if num_rows % batch_size else num_rows // batch_size				,      unit="""ba"""				,      disable=not logging.is_progress_bar_enabled()				,      desc="""Creating SQL from Arrow format"""				,      ):
											written += num_rows

					return written



