"""
def __init__(self, host: str | None = None,
- port: str | int | None = None,
- password: str | None = None):
+ port: str | int | None = None,
+ password: str | None = None):
#: Connection pool
self._pool = ConnectionPool(max_connections=CONNECTION_MAX)
#: connection timeout
"sendmessage": self._fetch_nothing,
}
self.command = None
- self._command_list = None
+ self._command_list: list | None = None
self.args = None
self.pool = pool
self.host = (server, port)
amount -= len(result)
return bytes(chunk)
- async def _read_line(self, binary=False):
+ async def _read_line(self):
line = await self.connection.readline()
line = line.decode('utf-8')
if not line.endswith('\n'):
return None
return line
- async def _read_pair(self, separator, binary=False):
- line = await self._read_line(binary=binary)
+ async def _read_pair(self, separator):
+ line = await self._read_line()
if line is None:
return None
pair = line.split(separator, 1)
raise MPDProtocolError(f"Could not parse pair: '{line}'")
return pair
- async def _read_pairs(self, separator=": ", binary=False):
- pair = await self._read_pair(separator, binary=binary)
+ async def _read_pairs(self, separator=": "):
+ pair = await self._read_pair(separator)
while pair:
yield pair
- pair = await self._read_pair(separator, binary=binary)
+ pair = await self._read_pair(separator)
async def _read_list(self):
seen = None
async def _fetch_composite(self):
obj = {}
- async for key, value in self._read_pairs(binary=True):
+ async for key, value in self._read_pairs():
key = key.lower()
obj[key] = value
if key == 'binary':
raise ConnectionError('Error reading binary content: '
f'Expects {amount}B, got {data_bytes}')
# Fetches trailing new line
- await self._read_line(binary=True)
+ await self._read_line()
#ALT: await self.connection.readuntil(b'\n')
# Fetches SUCCESS code
- await self._read_line(binary=True)
+ await self._read_line()
#ALT: await self.connection.readuntil(b'OK\n')
return obj