socket.socket

All Samples(6984) | Call(6726) | Derive(56) | Import(202)

socket([family[, type[, proto]]]) -> socket object

Open a socket of the given type.  The family argument specifies the
address family; it defaults to AF_INET.  The type argument specifies
whether this is a stream (SOCK_STREAM, this is the default)
or datagram (SOCK_DGRAM) socket.  The protocol argument defaults to 0,
specifying the default protocol.  Keyword arguments are accepted.

A socket object represents one endpoint of a network connection.

Methods of socket objects (keyword arguments not allowed):

accept() -- accept a connection, returning new socket and client address
bind(addr) -- bind the socket to a local address
close() -- close the socket
connect(addr) -- connect the socket to a remote address
connect_ex(addr) -- connect, return an error code instead of an exception
dup() -- return a new socket object identical to the current one [*]
fileno() -- return underlying file descriptor
getpeername() -- return remote address [*]
getsockname() -- return local address
getsockopt(level, optname[, buflen]) -- get socket options
gettimeout() -- return timeout or None
listen(n) -- start listening for incoming connections
makefile([mode, [bufsize]]) -- return a file object for the socket [*]
recv(buflen[, flags]) -- receive data
recv_into(buffer[, nbytes[, flags]]) -- receive data (into a buffer)
recvfrom(buflen[, flags]) -- receive data and sender's address
recvfrom_into(buffer[, nbytes, [, flags])
  -- receive data and sender's address (into a buffer)
sendall(data[, flags]) -- send all data
send(data[, flags]) -- send data, may not send all of it
sendto(data[, flags], addr) -- send data to a given address
setblocking(0 | 1) -- set or clear the blocking I/O flag
setsockopt(level, optname, value) -- set socket options
settimeout(None | float) -- set or clear the timeout
shutdown(how) -- shut down traffic in one or both directions

 [*] not available on all platforms!

src/p/y/pylibssh2-HEAD/examples/ssh_x11.py pylibssh2(Download)

import os
import select, sys
import subprocess
from socket import socket, AF_INET, AF_UNIX, SOCK_STREAM, SHUT_RDWR
import tty, termios
 
import libssh2

def x11_callback(session, channel, shost, sport, abstract):
    display = os.environ["DISPLAY"]
    display_port = display[display.index(":")+1]
    _path_unix_x = "/tmp/.X11-unix/X%s" % display_port
    if display[:5] == "unix:" or display[0] == ':':
        x11_sock = socket(AF_UNIX, SOCK_STREAM)
        x11_sock.connect(_path_unix_x)

    password = sys.argv[3]
    port = int(sys.argv[4])
 
    sock = socket(AF_INET, SOCK_STREAM)
    try:
        sock.connect((hostname, port))
        sock.setblocking(1)

src/p/y/pylibssh2-1.0.0/examples/ssh_x11.py pylibssh2(Download)

import fcntl
import os
import select, signal, struct, sys
from socket import socket, AF_INET, AF_UNIX, SOCK_STREAM, SHUT_RDWR
import tty, termios
 
import libssh2

def x11_callback(session, channel, shost, sport, abstract):
    display = os.environ["DISPLAY"]
    display_port = display[display.index(":")+1]
    _path_unix_x = "/tmp/.X11-unix/X%s" % display_port
    if display[:5] == "unix:" or display[0] == ':':
        x11_sock = socket(AF_UNIX, SOCK_STREAM)
        x11_sock.connect(_path_unix_x)

    password = sys.argv[3]
    port = int(sys.argv[4])
 
    sock = socket(AF_INET, SOCK_STREAM)
    try:
        sock.connect((hostname, port))
        sock.setblocking(1)

src/g/r/graphite-web-0.9.6/examples/example-client.py graphite-web(Download)

import os
import platform 
import subprocess
from socket import socket
 
CARBON_SERVER = '127.0.0.1'
CARBON_PORT = 2003

    length = len(output)
    return output[length - 3:length]
 
sock = socket()
try:
  sock.connect( (CARBON_SERVER,CARBON_PORT) )
except:

src/c/o/collective.solr-1.0/src/collective/solr/timeout.py collective.solr(Download)

from urllib2 import build_opener, HTTPHandler
from httplib import HTTPConnection
from socket import getaddrinfo, socket, error, SOCK_STREAM
 
 
class HTTPConnectionWithTimeout(HTTPConnection):

    def connect(self):
        """ copied from httplib.py and added timeout handling """
        msg = "getaddrinfo returns an empty list"
        for res in getaddrinfo(self.host, self.port, 0, SOCK_STREAM):
            af, socktype, proto, canonname, sa = res
            try:
                self.sock = socket(af, socktype, proto)

src/f/u/fusil-1.3.2/fusil/network/server.py fusil(Download)

from fusil.project_agent import ProjectAgent
from fusil.network.server_client import ServerClient
from fusil.network.tools import formatAddress
from socket import (socket, error as socket_error,
    AF_INET, SOCK_STREAM, SOL_SOCKET, SO_REUSEADDR)
from ptrace.error import writeError
from select import select

    def bind(self, address, family=AF_INET, type=SOCK_STREAM, reuse_address=True):
        try:
            self.socket = socket(family, type)
            if reuse_address:
                self.socket.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
            self.family = family
            self.socket.bind(address)

src/p/y/Pydev-HEAD/plugins/org.python.pydev.debug/pysrc/pydevd_comm.py Pydev(Download)

    import Queue as PydevQueue
except ImportError:
    import queue as PydevQueue
from socket import socket
from socket import AF_INET, SOCK_STREAM
try:
    from urllib import quote

def StartServer(port):
    """ binds to a port, waits for the debugger to connect """
    s = socket(AF_INET, SOCK_STREAM)
    s.bind(('', port))
    s.listen(1)
    newSock, _addr = s.accept()
    return newSock

def StartClient(host, port):
    """ connects to a host/port """
    PydevdLog(1, "Connecting to ", host, ":", str(port))
    try:
        s = socket(AF_INET, SOCK_STREAM);
 
        s.connect((host, port))

src/d/y/dyna-gen-HEAD/code/releases/dynagen-0.3.2/dynamips_lib.py dyna-gen(Download)

"""
 
import sys, os
from socket import socket, timeout, AF_INET, SOCK_STREAM
 
version = "0.3.4.061406"
NOSEND = False       # Disable sending any commands to the back end for debugging

    def __init__(self, host, port=7200, timeout=300):
        self.s = socket(AF_INET, SOCK_STREAM)
        self.s.setblocking(0)
        self.s.settimeout(timeout)
        if not NOSEND:
            try:
                self.s.connect((host,port))

src/n/e/neitris-HEAD/neitris/trunk/neitris_server.py neitris(Download)

import struct
import sys, time, pygame
from select import select
from socket import socket, AF_INET, SOCK_STREAM, SOL_SOCKET, SO_REUSEADDR
import neitris_utils
import neitris_cfg

 
# Create Reading Sockets:
for PORT in PORTS:
    portsock = socket(AF_INET, SOCK_STREAM)
    portsock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    portsock.bind((HOSTNAME, PORT))

src/c/l/clamwin-HEAD/trunk/addons/pyc/cwclient.py clamwin(Download)

# for more details.
# ======================================================================
 
from socket import socket, AF_INET, SOCK_STREAM
from os import getcwd
from os.path import join as path_join
 
EICAR='*H+H$!ELIF-TSET-SURIVITNA-DRADNATS-RACIE$}7)CC7)^P(45XZP\\4[PA@%P!O5X'[::-1]
 
def connect():
    s = socket(AF_INET, SOCK_STREAM)

 
    f.write('STREAM\n')
    port = int(f.readline().strip().split('PORT ', 1).pop())
    s = socket(AF_INET, SOCK_STREAM)
    s.connect(('localhost', port))
    s.send(EICAR)
    s.close()
    print f.readline().strip()
 
    f.write('STREAM\n')
    port = int(f.readline().strip().split('PORT ', 1).pop())
    s = socket(AF_INET, SOCK_STREAM)

src/c/a/caslib-0.3.1/caslib/validating_https.py caslib(Download)

 
'''Validate certificates and hostnames on https connections with :mod:`urllib2` and :mod:`httplib`'''
 
from socket import socket, SOCK_STREAM, getaddrinfo, error as SocketError
from httplib import HTTPSConnection
import urllib2

 
	    try:
		_log.debug('ValidatingHTTPSConnection to %r', addr)
		ssl = SSLSocket(socket(af, socktype, proto),
		    keyfile=self.key_file, certfile=self.cert_file,
		    cert_reqs=self.cert_reqs, ca_certs=self.ca_certs,
		    ssl_version=self.ssl_version)

1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 Next