Reference

Reference material consisting of package details.

Wordly Utility classes for use in your python scripts.

Word

Bases: UserString

str subclass that provides an interface for retrieving definitions of terms.

Source code in src/wordly/words.py

class Word(UserString):
    """`str` subclass that provides an interface for retrieving definitions of terms."""

    def __init__(self, seq, hostname: str = "dict.org", port: int = 2628, client=None):
        """Initialize."""
        super().__init__(seq)

        self.client = client or DictClient(hostname=hostname, port=port)
        self._cache = {}

    @property
    async def adefinition(self) -> str | None:
        """Return definition of word."""
        if not self._cache:
            data = await self.client.define(self.data)
            self._cache.update(data.mapping)
        if definition := self._cache.get(Status.DEFINITION.name):
            return definition.decode()

    @property
    def definition(self) -> str | None:
        """Return definition of word."""
        if not self._cache:
            loop = asyncio.get_event_loop()
            data = loop.run_until_complete(self.client.define(self.data))
            self._cache.update(data.mapping)
        if definition := self._cache.get(Status.DEFINITION.name):
            return definition.decode()

adefinition async property

Return definition of word.

definition property

Return definition of word.

__init__(seq, hostname='dict.org', port=2628, client=None)

Initialize.

Source code in src/wordly/words.py

def __init__(self, seq, hostname: str = "dict.org", port: int = 2628, client=None):
    """Initialize."""
    super().__init__(seq)

    self.client = client or DictClient(hostname=hostname, port=port)
    self._cache = {}

DICT client.

DictClient

Client.

Source code in src/wordly/client.py

class DictClient:
    """Client."""

    def __init__(self, hostname: str = "dict.org", port: int = 2628) -> None:
        """Initialize."""
        self.hostname = hostname
        self.port = port
        self.line_reader = DictParser()
        self.parsers = [self.line_reader]
        self.reader: asyncio.StreamReader | None = None
        self.writer: asyncio.StreamWriter | None = None
        self.READ_BYTES = 1024

    def __repr__(self) -> str:
        """Return string representation of object."""
        return f"{self.__class__.__name__}({self.hostname=}, {self.port=})"

    async def __aenter__(self):
        """Enter method for async context manager."""
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        """Exit method for async context manager."""
        await self.disconnect()

    async def connect(self):
        """Upon successful connection a status code of 220 is expected."""
        self.reader, self.writer = await asyncio.open_connection(
            self.hostname, self.port
        )
        connection_info = await self.reader.read(self.READ_BYTES)
        self.line_reader.feed(connection_info)

        if Status.INITIAL_CONNECTION.name in self.line_reader.mapping:
            return self.reader, self.writer

        raise ConnectionError(f"Could not connect to: {self.hostname=}, {self.port=}")

    async def disconnect(self):
        """Close client connection."""
        self.writer.write(b"QUIT\r\n")
        await self.writer.drain()
        while Status.CLOSING_CONNECTION.name not in self.line_reader.mapping:
            closing_data = await self.reader.read(self.READ_BYTES)
            for line_reader in self.parsers:
                line_reader.feed(closing_data)
        self.writer.close()
        await self.writer.wait_closed()

    async def _send(self, command: bytes) -> DictParser:
        """Return line reader given a command."""
        if None in (self.reader, self.writer):
            self.reader, self.writer = await self.connect()
        else:
            new_line_reader = DictParser()
            new_line_reader.mapping[Status.INITIAL_CONNECTION.name] = (
                self.line_reader.mapping[Status.INITIAL_CONNECTION.name]
            )
            self.line_reader = new_line_reader
            self.parsers.append(self.line_reader)

        self.writer.write(command)
        await self.writer.drain()

        while (
            Status.COMMAND_COMPLETE.name not in self.line_reader.mapping
            and Status.NO_MATCH.name not in self.line_reader.mapping
        ):
            command_data = await self.reader.read(self.READ_BYTES)
            self.line_reader.feed(command_data)
        return self.line_reader

    async def define(self, word: str, database: str = "!") -> DictParser:
        """Return line reader given word and database."""
        command = f"DEFINE {database} {word}\r\n".encode()
        return await self._send(command)

    async def help(self) -> DictParser:
        """Return line reader with helpful information."""
        command = b"HELP\r\n"
        return await self._send(command)

    async def match(
        self, word: str, database: str = "*", strategy: str = "."
    ) -> DictParser:
        """Match a word in a database using a strategy."""
        command = f"MATCH {database} {strategy} {word}".encode()
        return await self._send(command)

    async def show(self, option: str = "DB") -> DictParser:
        """Show more information."""
        command = f"SHOW {option}".encode()
        return await self._send(command)

__aenter__() async

Enter method for async context manager.

Source code in src/wordly/client.py

async def __aenter__(self):
    """Enter method for async context manager."""
    await self.connect()
    return self

__aexit__(exc_type, exc, tb) async

Exit method for async context manager.

Source code in src/wordly/client.py

async def __aexit__(self, exc_type, exc, tb):
    """Exit method for async context manager."""
    await self.disconnect()

__init__(hostname='dict.org', port=2628)

Initialize.

Source code in src/wordly/client.py

def __init__(self, hostname: str = "dict.org", port: int = 2628) -> None:
    """Initialize."""
    self.hostname = hostname
    self.port = port
    self.line_reader = DictParser()
    self.parsers = [self.line_reader]
    self.reader: asyncio.StreamReader | None = None
    self.writer: asyncio.StreamWriter | None = None
    self.READ_BYTES = 1024

__repr__()

Return string representation of object.

Source code in src/wordly/client.py

def __repr__(self) -> str:
    """Return string representation of object."""
    return f"{self.__class__.__name__}({self.hostname=}, {self.port=})"

connect() async

Upon successful connection a status code of 220 is expected.

Source code in src/wordly/client.py

async def connect(self):
    """Upon successful connection a status code of 220 is expected."""
    self.reader, self.writer = await asyncio.open_connection(
        self.hostname, self.port
    )
    connection_info = await self.reader.read(self.READ_BYTES)
    self.line_reader.feed(connection_info)

    if Status.INITIAL_CONNECTION.name in self.line_reader.mapping:
        return self.reader, self.writer

    raise ConnectionError(f"Could not connect to: {self.hostname=}, {self.port=}")

define(word, database='!') async

Return line reader given word and database.

Source code in src/wordly/client.py

async def define(self, word: str, database: str = "!") -> DictParser:
    """Return line reader given word and database."""
    command = f"DEFINE {database} {word}\r\n".encode()
    return await self._send(command)

disconnect() async

Close client connection.

Source code in src/wordly/client.py

async def disconnect(self):
    """Close client connection."""
    self.writer.write(b"QUIT\r\n")
    await self.writer.drain()
    while Status.CLOSING_CONNECTION.name not in self.line_reader.mapping:
        closing_data = await self.reader.read(self.READ_BYTES)
        for line_reader in self.parsers:
            line_reader.feed(closing_data)
    self.writer.close()
    await self.writer.wait_closed()

help() async

Return line reader with helpful information.

Source code in src/wordly/client.py

async def help(self) -> DictParser:
    """Return line reader with helpful information."""
    command = b"HELP\r\n"
    return await self._send(command)

match(word, database='*', strategy='.') async

Match a word in a database using a strategy.

Source code in src/wordly/client.py

async def match(
    self, word: str, database: str = "*", strategy: str = "."
) -> DictParser:
    """Match a word in a database using a strategy."""
    command = f"MATCH {database} {strategy} {word}".encode()
    return await self._send(command)

show(option='DB') async

Show more information.

Source code in src/wordly/client.py

async def show(self, option: str = "DB") -> DictParser:
    """Show more information."""
    command = f"SHOW {option}".encode()
    return await self._send(command)

DICT status codes.

BytesEnum

Bases: bytes, ReprEnum

Enum where members are also (and must be) bytes.

Source code in src/wordly/status_codes.py

class BytesEnum(bytes, ReprEnum):
    """Enum where members are also (and must be) `bytes`."""

    def __new__(cls, *values):
        """Values must be already of type `bytes`."""
        if any(not isinstance(value, bytes) for value in values):
            raise TypeError(f"All values must be of type `bytes`: got {values}")
        return super().__new__(cls, *values)

__new__(*values)

Values must be already of type bytes.

Source code in src/wordly/status_codes.py

def __new__(cls, *values):
    """Values must be already of type `bytes`."""
    if any(not isinstance(value, bytes) for value in values):
        raise TypeError(f"All values must be of type `bytes`: got {values}")
    return super().__new__(cls, *values)

Status

Bases: BytesEnum

Enumeration of possible status responses from a DICT server.

Note

The first digit of the response has the following meaning:

1yz - Positive Preliminary reply
2yz - Positive Completion reply
3yz - Positive Intermediate reply
4yz - Transient Negative Completion reply
5yz - Permanent Negative Completion reply

The next digit in the code indicates the response category:

x0z - Syntax
x1z - Information (e.g., help)
x2z - Connections
x3z - Authentication
x4z - Unspecified as yet
x5z - DICT System (These replies indicate the status of the
      receiver DICT system vis-a-vis the requested transfer
      or other DICT system action.)
x8z - Nonstandard (private implementation) extensions

Source code in src/wordly/status_codes.py

@unique
class Status(BytesEnum):
    """Enumeration of possible status responses from a DICT server.

    Note:
        The first digit of the response has the following meaning:

            1yz - Positive Preliminary reply
            2yz - Positive Completion reply
            3yz - Positive Intermediate reply
            4yz - Transient Negative Completion reply
            5yz - Permanent Negative Completion reply

        The next digit in the code indicates the response category:

            x0z - Syntax
            x1z - Information (e.g., help)
            x2z - Connections
            x3z - Authentication
            x4z - Unspecified as yet
            x5z - DICT System (These replies indicate the status of the
                  receiver DICT system vis-a-vis the requested transfer
                  or other DICT system action.)
            x8z - Nonstandard (private implementation) extensions

    """

    DATABASE_PRESENT = b"110"
    STRATEGIES_AVAILABLE = b"111"
    DATABASE_INFO = b"112"
    HELP = b"113"
    SERVER_INFO = b"114"
    DEFINITION_RETRIEVED = b"150"
    DEFINITION = b"151"
    MATCHES_FOUND = b"152"
    INITIAL_CONNECTION = b"220"
    CLOSING_CONNECTION = b"221"
    AUTH_SUCCESSFUL = b"230"
    COMMAND_COMPLETE = b"250"
    SERVER_TEMPORARILY_UNAVAILABLE = b"420"
    SERVER_SHUTTING_DOWN = b"421"
    COMMAND_NOT_RECOGNIZED = b"500"
    ILLEGAL_PARAMETERS = b"501"
    COMMAND_NOT_IMPLEMENTED = b"502"
    COMMAND_PARAMETER_NOT_IMPLEMENTED = b"503"
    ACCESS_DENIED = b"530"
    NO_MATCH = b"552"

    @classmethod
    @cache
    def statuses(cls) -> dict[bytes, Status]:
        """Return `dict` of status values as keys and members as values."""
        return {member.value: member for _, member in cls.__members__.items()}

    @classmethod
    def by_value(cls, value: bytes) -> Status | None:
        """Return a status code given a `bytes` value."""
        return cls.statuses().get(value)

by_value(value) classmethod

Return a status code given a bytes value.

Source code in src/wordly/status_codes.py

@classmethod
def by_value(cls, value: bytes) -> Status | None:
    """Return a status code given a `bytes` value."""
    return cls.statuses().get(value)

statuses() cached classmethod

Return dict of status values as keys and members as values.

Source code in src/wordly/status_codes.py

@classmethod
@cache
def statuses(cls) -> dict[bytes, Status]:
    """Return `dict` of status values as keys and members as values."""
    return {member.value: member for _, member in cls.__members__.items()}

DICT parser or line reader.

DictParser

Line reader for parsing byte stream of DICT protocol.

Creates a map of DICT status code and associated information.

Source code in src/wordly/parser.py

class DictParser:
    """Line reader for parsing byte stream of DICT protocol.

    Creates a map of DICT status code and associated information.
    """

    def __init__(self, delimiter: bytes = b"\r\n"):
        """Initialize."""
        self.line = bytearray()
        self.mapping = defaultdict(bytearray)
        self.DELIMITER = delimiter

    def _process_line(self, ending: bytes = b""):
        """Process line."""
        code = self.line[:3]
        status = Status.by_value(bytes(code))

        if status:
            data = self.line[4:]
            self.part = status.name
        else:
            data = self.line

        buf = self.mapping[self.part]
        buf.extend(data)
        buf.extend(ending)

    def feed(self, stream: bytes):
        """Feed stream of `bytes` to line reader.

        Calls `_process_line` on bytes stream until delimiter
        can no longer be found.
        """
        split = stream.split(self.DELIMITER, 1)
        while len(split) > 1:
            old, new = split
            self.line += old
            self._process_line(b"\n")
            self.line = b""
            split = new.split(self.DELIMITER, 1)

        if line := split[0]:
            self.line += line
            self._process_line()
            self.line = b""

__init__(delimiter=b'\r\n')

Initialize.

Source code in src/wordly/parser.py

def __init__(self, delimiter: bytes = b"\r\n"):
    """Initialize."""
    self.line = bytearray()
    self.mapping = defaultdict(bytearray)
    self.DELIMITER = delimiter

feed(stream)

Feed stream of bytes to line reader.

Calls _process_line on bytes stream until delimiter can no longer be found.

Source code in src/wordly/parser.py

def feed(self, stream: bytes):
    """Feed stream of `bytes` to line reader.

    Calls `_process_line` on bytes stream until delimiter
    can no longer be found.
    """
    split = stream.split(self.DELIMITER, 1)
    while len(split) > 1:
        old, new = split
        self.line += old
        self._process_line(b"\n")
        self.line = b""
        split = new.split(self.DELIMITER, 1)

    if line := split[0]:
        self.line += line
        self._process_line()
        self.line = b""