(self, server: Server, conn: AsyncConnection)
| 449 | return None |
| 450 | |
| 451 | def pin(self, server: Server, conn: AsyncConnection) -> None: |
| 452 | self.sharded = True |
| 453 | self.pinned_address = server.description.address |
| 454 | if server.description.server_type == SERVER_TYPE.LoadBalancer: |
| 455 | conn.pin_txn() |
| 456 | self.conn_mgr = _ConnectionManager(conn, False) |
| 457 | |
| 458 | async def unpin(self) -> None: |
| 459 | self.pinned_address = None |
no test coverage detected