Repository: redis/redis-py
File: redis/asyncio/connection.py (Python)

Code before:
        return cls(**kwargs)

    def __init__(
        self,
        connection_class: Type[AbstractConnection] = Connection,
        max_connections: Optional[int] = None,
        **connection_kwargs,
    ):
        max_connections = max_connections or 2**31
        if not isinstance(max_connections, int) or max_connections < 0:
            raise ValueError('"max_connections" must be a positive integer')

        self.connection_class = connection_class
        self.connection_kwargs = connection_kwargs
        self.max_connections = max_connections

        self._available_connections: List[AbstractConnection] = []
        self._in_use_connections: Set[AbstractConnection] = set()
        self.encoder_class = self.connection_kwargs.get("encoder_class", Encoder)
        self._lock = asyncio.Lock()
        self._event_dispatcher = self.connection_kwargs.get("event_dispatcher", None)
        if self._event_dispatcher is None:
            self._event_dispatcher = EventDispatcher()

    def __repr__(self):
        return (
            f"<{self.__class__.__module__}.{self.__class__.__name__}"
            f"({self.connection_class(**self.connection_kwargs)!r})>"
        )

    def reset(self):
        self._available_connections = []
        self._in_use_connections = weakref.WeakSet()

    def can_get_connection(self) -> bool:
        """Return True if a connection can be retrieved from the pool."""
        return (
            self._available_connections
            or len(self._in_use_connections) < self.max_connections
        )

    @deprecated_args(
        args_to_warn=["*"],
        reason="Use get_connection() without args instead",
        version="5.3.0",
    )
    async def get_connection(self, command_name=None, *keys, **options):
        async with self._lock:
            """Get a connected connection from the pool"""
            connection = self.get_available_connection()
            try:

Review comment (suggestion):
As with the synchronous version, consider ensuring a consistent order for connection_kwargs in the repr output if the order is significant for tests or logging; sorting keys or specifying an order can help avoid non-deterministic behavior.
```suggestion
        conn_kwargs = ",".join([f"{k}={v}" for k, v in sorted(self.connection_kwargs.items())])
```

Code after:
        return cls(**kwargs)

    def __init__(
        self,
        connection_class: Type[AbstractConnection] = Connection,
        max_connections: Optional[int] = None,
        **connection_kwargs,
    ):
        max_connections = max_connections or 2**31
        if not isinstance(max_connections, int) or max_connections < 0:
            raise ValueError('"max_connections" must be a positive integer')

        self.connection_class = connection_class
        self.connection_kwargs = connection_kwargs
        self.max_connections = max_connections

        self._available_connections: List[AbstractConnection] = []
        self._in_use_connections: Set[AbstractConnection] = set()
        self.encoder_class = self.connection_kwargs.get("encoder_class", Encoder)
        self._lock = asyncio.Lock()
        self._event_dispatcher = self.connection_kwargs.get("event_dispatcher", None)
        if self._event_dispatcher is None:
            self._event_dispatcher = EventDispatcher()

    def __repr__(self):
        conn_kwargs = ",".join([f"{k}={v}" for k, v in self.connection_kwargs.items()])
        return (
            f"<{self.__class__.__module__}.{self.__class__.__name__}"
            f"(<{self.connection_class.__module__}.{self.connection_class.__name__}"
            f"({conn_kwargs})>)>"
        )

    def reset(self):
        self._available_connections = []
        self._in_use_connections = weakref.WeakSet()

    def can_get_connection(self) -> bool:
        """Return True if a connection can be retrieved from the pool."""
        return (
            self._available_connections
            or len(self._in_use_connections) < self.max_connections
        )

    @deprecated_args(
        args_to_warn=["*"],
        reason="Use get_connection() without args instead",
        version="5.3.0",
    )
    async def get_connection(self, command_name=None, *keys, **options):
        async with self._lock:
            """Get a connected connection from the pool"""