Skip to main content

CDP-Socket

Feel free to test my code!

Getting Started

Dependencies

Installing

CDPSocket

from cdp_socket.utils.utils import launch_chrome, random_port
from cdp_socket.socket import CDPSocket

import os
import asyncio

async def main():
    PORT = random_port()
    process = launch_chrome(PORT)
    
    async with CDPSocket(PORT) as base_socket:
        targets = await base_socket.targets
        sock1 = await base_socket.get_socket(targets[0])
        targets = await sock1.exec("Target.getTargets")
        print(targets)
    os.kill(process.pid, 15)


asyncio.run(main())

Single socket

from cdp_socket.utils.utils import launch_chrome, random_port
from cdp_socket.utils.conn import get_websock_url

from cdp_socket.socket import SingleCDPSocket

import os
import asyncio


async def main():
    PORT = random_port()
    process = launch_chrome(PORT)

    websock_url = await get_websock_url(PORT, timeout=5)
    async with SingleCDPSocket(websock_url, timeout=5) as sock:
        targets = await sock.exec("Target.getTargets")
        print(targets)

    os.kill(process.pid, 15)


asyncio.run(main())

on_closed callback

from cdp_socket.socket import SingleCDPSocket

def on_closed(code, reason):
    print("Closed with: ", code, reason)

async with SingleCDPSocket(websock_url, timeout=5) as sock:
    sock.on_closed.append(on_closed)
    # close window for dispatching this event
    targets = await sock.exec("Target.getTargets")
    print(targets)

add event listener

from cdp_socket.socket import SingleCDPSocket
import asyncio

def on_detached(params):
    print("Detached with: ", params)
    
async with SingleCDPSocket(websock_url, timeout=5) as sock:
        # close window for dispatching this event
        sock.add_listener('Inspector.detached', on_detached)
        await asyncio.sleep(1000)

iterate over event

from cdp_socket.socket import SingleCDPSocket

async with SingleCDPSocket(websock_url, timeout=5) as sock:
    async for i in sock.method_iterator('Inspector.detached'):
        print(i)
        break

wait for event

from cdp_socket.socket import SingleCDPSocket

async with SingleCDPSocket(websock_url, timeout=5) as sock:
    res = await sock.wait_for('Inspector.detached')
    print(res)

synchronous

from cdp_socket.utils.utils import launch_chrome, random_port
from cdp_socket.utils.conn import get_websock_url

from cdp_socket.socket import SingleCDPSocket

import os
import asyncio

PORT = random_port()
process = launch_chrome(PORT)

loop = asyncio.get_event_loop()
websock_url = loop.run_until_complete(get_websock_url(PORT, timeout=5))

conn = loop.run_until_complete(SingleCDPSocket(websock_url, timeout=5, loop=loop))
targets = loop.run_until_complete(conn.exec("Target.getTargets"))
print(targets)

os.kill(process.pid, 15)

Help

Please feel free to open an issue or fork!

Todo

Deprecated

Authors

Aurin Aegerter

License

Shield: CC BY-NC-SA 4.0

This work is licensed under a Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International License.

CC BY-NC-SA 4.0

Disclaimer

I am not responsible what you use the code for!!! Also no warranty!

Acknowledgments

Inspiration, code snippets, etc.

contributors

  • thanks to @Redrrx who gave me some starting-points

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

cdp-socket-1.0.2.tar.gz (9.3 kB view details)

Uploaded Source

File details

Details for the file cdp-socket-1.0.2.tar.gz.

File metadata

  • Download URL: cdp-socket-1.0.2.tar.gz
  • Upload date:
  • Size: 9.3 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/4.0.2 CPython/3.8.10

File hashes

Hashes for cdp-socket-1.0.2.tar.gz
Algorithm Hash digest
SHA256 a9e2cd9a627394ee990b2283b59c88e42a387d4d92d1952a0171572d114c2955
MD5 695adea0f5380684cf1e9a20f1f1d0ae
BLAKE2b-256 bb20888a1944953881059799d8d7fd25d7fcdf3a94ce1462e52a80c0499c936f

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page