Python API

The methods and classes documented here only comprise a portion of all of the functionality of the lmk package, but these were chosen based on the pieces most likely to be useful to users of the package. You may not have any need for using the python API directly at all, as it's not required. It is expected that most users will mainly use the Jupyter and CLI integrations to monitor jupyter notebooks and command line processes respectively, in which case there's no need to call any of these functions directly.

Top-level API functions

The following methods are available at the top level of the lmk module, and allow you to perform basic operations with LMK manually such as authenticating and sending notifications.

`notify`

def notify(message: str,
           content_type: str = "text/markdown",
           notification_channels: Optional[List[Union[
               str, NotificationChannelResponse]]] = None,
           notify: bool = True,
           async_req: bool = False) -> EventResponse

Send a notification to one of your configured notification channels.

Note: This method requires you to be logged in to LMK.

Usage Example

import lmk

# Check that the client is logged in (optional)
# If using for the first time on a new device, call lmk.login()
assert lmk.logged_in()

lmk.notify("Hello, world!")

Arguments:

message (str): The content of the notification you want to send
content_type (str, optional): The MIME type of the message you want to send; text/plain and text/markdown are supported. Defaults to text/markdown
notification_channels (List[str | NotificationChannelResponse], optional): A list of notification channel IDs or notification channel objects that you want to send the notification to. If None and notify = True (the default), this will be sent to the default notification channel for your account, which is the primary email address associated with your account by default. Defaults to None
notify (bool, optional): True if you want to send a notification to one or more of your configured notification channels. If False, this notification will only be visible via the LMK web app. Defaults to True
async_req (bool, optional): True if you want to send the request asynchronously, in which case this method will return a coroutine. Defaults to False.

Returns:

EventResponse: The event object corresponding to the sent notification

`logged_in`

def logged_in() -> bool

Check whether the client is currently logged in

Returns:

bool: a boolean indicating if you are logged in or not

`login`

def login(scope: Optional[str] = None,
          timeout: float = 300.0,
          poll_interval: float = 1.0,
          auth_mode: Optional[str] = None,
          force: bool = False,
          print_function: Optional[Callable[[str], None]] = None) -> None

Authenticate LMK interactively. If possible, this will open

a new auth session in a web browser. If not, it will print the auth URL and you will have to navigate to it in a browser manually

Arguments:

auth_mode (str, optional): jupyter, browser or manual. If None, the first of those three that is available in the current context will be chosen automatically.
scope (str, optional): If desired, you may manually specify scopes that the retrieved access token should have, separated by spaces. If None, all scopes will be requested. The user may modify these scopes during the OAuth flow to make them more restrictive.
timeout (float, optional): A float indicating how long the client should wait for authentication to succeed before considering the authentication attempt failed.
poll_interval (float, optional): The poll interval for checking if the created authentication session has succeeded.
force (bool, optional): By default if you are already logged in, this method will simply return immediately. By passing force=True, you may force the client to replace the current authentication information and log in again.
print_function (Callable[[str], None], optional): Pass a print function to use when printing the "Log in successful" message after successful login. By default this will use the built-in print function.

Returns:

None: This method does not return anything

`logout`

def logout() -> None

Get rid of the current access token. After calling this, you will

need to log in again in order to use methods that require authentication such as notify()

Returns:

None: This method does not return anything

`list_notification_channels`

def list_notification_channels(
        async_req: bool = False) -> List[NotificationChannelResponse]

List the notification channels that are available to send notifications to

Arguments:

async_req (bool, optional): True if you want to send the request asynchronously, in which case this method will return a coroutine. Defaults to False.

Returns:

List[NotificationChannelResponse]: A list of notification channels that the current client has access to.

`create_session`

def create_session(name: str,
                   state: Optional[Union[Dict[str, Any], ProcessSessionState,
                                         JupyterSessionState]] = None,
                   async_req: bool = False) -> SessionResponse

Create an interactive session, which you can use to remotely monitor a process

or Jupyter Notebook remotely via the LMK web app. You shouldn't have to use this method directly in normal usage, rather it will be invoed by

Note: This method requires you to be logged in to LMK.

Arguments:

name (str): The name of the session. This will appear in the LMK app.
state (Dict[str, Any]): The initial state parameters for the session. The type field is always required, but the rest of the required fields depend on what type of session it is. See the REST API documentation for more information.
async_req (bool, optional): True if you want to send the request asynchronously, in which case this method will return a coroutine. Defaults to False.

Returns:

SessionResponse: The session object corresponding to the created session.

`session_connect`

@contextlib.asynccontextmanager
async def session_connect(
        session_id: str,
        read_only: bool = True) -> AsyncGenerator[WebSocket, None]

Connect via a web socket to an interactive session. This allows you to send state

updates to the session via a web socket, and receive remote state updates initiated through the LMK web app or API calls from other clients.

Arguments:

session_id (str): the ID of a previously created session that has not been ended yet.
read_only (bool, optional): Indicate whether to connect in "read only" mode. This means that updates cannot be sent via the web socket, only received. Defaults to True

Returns:

AsyncContextManager[WebSocket]: An asynchronous context manager yielding a WebSocket object

`end_session`

def end_session(session_id: str, async_req: bool = False) -> None

End an interactive session. After the session has been ended, its state cannot

be updated any more. Only call this when completely finished with using a session.

Arguments:

session_id (str): The ID of a previously created session that has not been ended yet
async_req (bool, optional): True if you want to send the request asynchronously, in which case this method will return a coroutine. Defaults to False.

Returns:

None: This method does not return anything

Class `Channels`

class Channels()

An instance of this class is available at the top level of the lmk module under the lmk.channels attribute. This class's function is to simplify fetching and searching for notification channels. Channels are always fetched lazily when required based on get(), list(), or iterating through the channels object (synchronously or asynchronously).

Usage Example

import lmk

# Check that the client is logged in (optional)
# If using for the first time on a new device, call lmk.login()
assert lmk.logged_in()

# Get an email notification channel
channel = lmk.channels.get(type="email")
# Get a text message notification channel asynchronously
channel = await lmk.channels.get(type="text-message", async_req=True)

# Iterate through notification channels
for channel in lmk.channels:
    print(channel)

# Iterate through notification channels asynchronously
async for channel in lmk.channels:
    print(channel)

`default`

@property
def default() -> Optional[str]

The default notification channel; If notify() is called without passing notification_channels, the notification will be sent to this channel. If you do not set this value, it will be set to the default notification channel for your account.

Usage Example

import lmk

lmk.channels.default = lmk.channels.get(type="text-message")

`fetch`

def fetch(async_req: bool = False, force: bool = False) -> None

Fetch notification channels.

Note: This requires the client to be logged in.

Arguments:

async_req (bool, optional): True if you want to send the request asynchronously, in which case this method will return a coroutine. Defaults to False.
force: If True, refetch notification channels even if they have already been fetched successfully. By default, fetching will be skipped if it's already been done successfully.

Returns:

None: This method does not return anything

`list`

def list(name: Optional[str] = None,
         type: Optional[Union[str, ChannelType]] = None,
         name_exact: bool = False,
         fetch: bool = True,
         async_req: bool = False) -> List[NotificationChannelResponse]

List notification channels. This will fetch notification channels if they haven't been

fetched yet.

Arguments:

name (str, optional): Filter down to notification channels with the given name
name_exact (bool, optional): Only return channels whose names exactly match name. By default, the name matching is case insensitive.
type (str | ChannelType, optional): Filter to only notification channels of this type
fetch (bool, optional): Fetch notification channels if they haven't been fetched yet.
async_req (bool, optional): True if you want to send the request asynchronously, in which case this method will return a coroutine. Defaults to False.

Returns:

List[NotificationChannelResponse]: A list of notification channels matching the given parameters

`get`

def get(name: Optional[str] = None,
        type: Optional[Union[str, ChannelType]] = None,
        name_exact: bool = False,
        fetch: bool = True,
        async_req: bool = False) -> Optional[NotificationChannelResponse]

Get a single notification channel with the given parameters.

Arguments:

name (str, optional): Filter down to notification channels with the given name
name_exact (bool, optional): Only return channels whose names exactly match name. By default, the name matching is case insensitive.
type (str | ChannelType, optional): Filter to only notification channels of this type
fetch (bool, optional): Fetch notification channels if they haven't been fetched yet.
async_req (bool, optional): True if you want to send the request asynchronously, in which case this method will return a coroutine. Defaults to False.

Returns:

NotificationChannelResponse | None: A notification channel matching the given parameters, or None if none exists.

Class `WebSocket`

class WebSocket()

Wrapper for aiohttp's web socket interface that supports reconnecting on failures

`send`

async def send(data: Any) -> None

Send a message to the web socket

Arguments:

data (Any): The data to send to the web socket. This must be JSON serializable.

Returns:

None: This method does not return anything

`aiter`

async def __aiter__()

Iterate asynchronously through messages received by the web socket.

Returns:

AsyncIterator[Any]: An asynchronous iterator of messages received from the web socket.

Python API

Top-level API functions​

notify​

logged_in​

login​

logout​

list_notification_channels​

create_session​

session_connect​

end_session​

Class Channels​

default​

fetch​

list​

get​

Class WebSocket​

send​

__aiter__​

Top-level API functions

`notify`

`logged_in`

`login`

`logout`

`list_notification_channels`

`create_session`

`session_connect`

`end_session`

Class `Channels`

`default`

`fetch`

`list`

`get`

Class `WebSocket`

`send`

`aiter`