examples: add a set of initial examples

Author: sumnerevans

README.md

@@ -1,9 +1,16 @@
 # LinkedIn Messaging API
 
+[![Python](https://github.com/sumnerevans/linkedin-messaging-api/actions/workflows/python.yaml/badge.svg)](https://github.com/sumnerevans/linkedin-messaging-api/actions/workflows/python.yaml)
+[![Matrix Chat](https://img.shields.io/matrix/linkedin-matrix:nevarro.space?server_fqdn=matrix.nevarro.space)](https://matrix.to/#/#linkedin-matrix:nevarro.space?via=nevarro.space&via=sumnerevans.com)
+
 An unofficial API for interacting with LinkedIn Messaging.
 
 Built using [aiohttp](https://docs.aiohttp.org/en/stable/).
 
+## Documentation
+
+See [`examples` directory](./examples).
+
 ## Credits
 
 Authentication technique from [@everping](https://github.com/everping) in the

examples/get-conversation.py

@@ -0,0 +1,44 @@
+from pathlib import Path
+
+import asyncio
+from linkedin_messaging import LinkedInMessaging, ChallengeException
+from linkedin_messaging.api_objects import URN
+
+cookie_path = Path(__file__).parent.joinpath("cookies.pickle")
+urn = URN(
+    "urn:li:fs_conversation:2-OTNkODIyYTEtODFjZS00NTdlLThlYTItYWQyMDg2NTc4YWMyXzAxMA=="
+)
+
+
+async def main():
+    linkedin = LinkedInMessaging()
+    if cookie_path.exists():
+        with open(cookie_path, "rb") as cf:
+            linkedin = LinkedInMessaging.from_pickle(cf.read())
+
+    if not await linkedin.logged_in():
+        try:
+            await linkedin.login("EMAIL", "PASSWORD")
+        except ChallengeException:
+            await linkedin.enter_2fa(input("2fa code: "))
+
+        with open(cookie_path, "wb+") as cf:
+            cf.write(linkedin.to_pickle())
+
+    # Get a list of all of the conversations for the given user.
+    conversations = await linkedin.get_conversations()
+    for c in conversations.elements:
+        print(c)
+
+    # Get a specific conversation by URN.
+    convo_resp = await linkedin.get_conversation(urn)
+    for element in convo_resp.elements:
+        print(element)
+
+    await linkedin.close()
+
+
+loop = asyncio.get_event_loop()
+asyncio.ensure_future(main())
+loop.run_forever()
+loop.close()

examples/listen-for-events.py

@@ -0,0 +1,48 @@
+import logging
+from pathlib import Path
+
+import asyncio
+from linkedin_messaging import LinkedInMessaging, ChallengeException
+
+cookie_path = Path(__file__).parent.joinpath("cookies.pickle")
+
+
+async def main():
+    logging.basicConfig()
+    logging.getLogger().setLevel(logging.DEBUG)
+
+    linkedin = LinkedInMessaging()
+    if cookie_path.exists():
+        with open(cookie_path, "rb") as cf:
+            linkedin = LinkedInMessaging.from_pickle(cf.read())
+
+    if not await linkedin.logged_in():
+        try:
+            await linkedin.login("EMAIL", "PASSWORD")
+        except ChallengeException:
+            await linkedin.enter_2fa(input("2fa code: "))
+
+        with open(cookie_path, "wb+") as cf:
+            cf.write(linkedin.to_pickle())
+
+    async def on_event(event):
+        print(event)
+
+    linkedin.add_event_listener("event", on_event)
+    linkedin.add_event_listener("reactionSummary", on_event)
+    linkedin.add_event_listener("reactionAdded", on_event)
+
+    task = asyncio.create_task(linkedin.start_listener())
+
+    # wait basically forever
+    await asyncio.sleep(2 ** 128)
+
+    asyncio.gather(task)
+
+    await linkedin.close()
+
+
+loop = asyncio.get_event_loop()
+asyncio.ensure_future(main())
+loop.run_forever()
+loop.close()

examples/pickle.py

@@ -0,0 +1,32 @@
+from pathlib import Path
+
+import asyncio
+from linkedin_messaging import LinkedInMessaging, ChallengeException
+
+cookie_path = Path(__file__).parent.joinpath("cookies.pickle")
+
+
+async def main():
+    linkedin = LinkedInMessaging()
+    if cookie_path.exists():
+        with open(cookie_path, "rb") as cf:
+            linkedin = LinkedInMessaging.from_pickle(cf.read())
+
+    if not await linkedin.logged_in():
+        try:
+            await linkedin.login("EMAIL", "PASSWORD")
+        except ChallengeException:
+            await linkedin.enter_2fa(input("2fa code: "))
+
+        with open(cookie_path, "wb+") as cf:
+            cf.write(linkedin.to_pickle())
+
+    print(await linkedin.get_user_profile())
+
+    await linkedin.close()
+
+
+loop = asyncio.get_event_loop()
+asyncio.ensure_future(main())
+loop.run_forever()
+loop.close()

examples/send-message.py

@@ -0,0 +1,49 @@
+import logging
+from pathlib import Path
+
+import asyncio
+from linkedin_messaging import LinkedInMessaging, ChallengeException
+from linkedin_messaging.api_objects import URN, MessageCreate, AttributedBody
+
+cookie_path = Path(__file__).parent.joinpath("cookies.pickle")
+urn = URN(
+    "urn:li:fs_conversation:2-OTNkODIyYTEtODFjZS00NTdlLThlYTItYWQyMDg2NTc4YWMyXzAxMA=="
+)
+
+
+async def main():
+    logging.basicConfig()
+    logging.getLogger().setLevel(logging.DEBUG)
+
+    linkedin = LinkedInMessaging()
+    if cookie_path.exists():
+        with open(cookie_path, "rb") as cf:
+            linkedin = LinkedInMessaging.from_pickle(cf.read())
+
+    if not await linkedin.logged_in():
+        try:
+            await linkedin.login("EMAIL", "PASSWORD")
+        except ChallengeException:
+            await linkedin.enter_2fa(input("2fa code: "))
+
+        with open(cookie_path, "wb+") as cf:
+            cf.write(linkedin.to_pickle())
+
+    # Send a simple message that has some text.
+    mc = MessageCreate(AttributedBody("test"))
+    print(await linkedin.send_message(urn, mc))
+
+    # Send a multimedia message.
+    with open("/path/to/the/cool-pic.jpg", "rb") as f:
+        attachment = await linkedin.upload_media(f.read(), "cool-pic.jpg", "image/jpeg")
+
+    mc = MessageCreate(AttributedBody(), attachments=[attachment])
+    print(await linkedin.send_message(urn, mc))
+
+    await linkedin.close()
+
+
+loop = asyncio.get_event_loop()
+asyncio.ensure_future(main())
+loop.run_forever()
+loop.close()

examples/simple-2fa.py

@@ -0,0 +1,25 @@
+from pathlib import Path
+
+import asyncio
+from linkedin_messaging import LinkedInMessaging, ChallengeException
+
+cookie_path = Path(__file__).parent.joinpath("cookies.pickle")
+
+
+async def main():
+    linkedin = LinkedInMessaging()
+
+    try:
+        await linkedin.login("EMAIL", "PASSWORD")
+    except ChallengeException:
+        await linkedin.enter_2fa(input("2fa code: "))
+
+    print(await linkedin.get_user_profile())
+
+    await linkedin.close()
+
+
+loop = asyncio.get_event_loop()
+asyncio.ensure_future(main())
+loop.run_forever()
+loop.close()