Network functionality

Network, HTTP, and URL functions

from fastcore.test import *
from nbdev.showdoc import *
from fastcore.nb_imports import *

URLs

---

source

urlquote

def urlquote(
    url
):

Update url’s path with urllib.parse.quote

urlquote("https://github.com/fastai/fastai/compare/master@{1.day.ago}…master")

'https://github.com/fastai/fastai/compare/master@%7B1.day.ago%7D%E2%80%A6master'

urlquote("https://www.google.com/search?q=你好")

'https://www.google.com/search?q=%E4%BD%A0%E5%A5%BD'

---

source

urlwrap

def urlwrap(
    url, data:NoneType=None, headers:NoneType=None
):

Wrap url in a urllib Request with urlquote

---

source

HTTP4xxClientError

def HTTP4xxClientError(
    url, code, msg, hdrs, fp
):

Base class for client exceptions (code 4xx) from url* functions

---

source

HTTP5xxServerError

def HTTP5xxServerError(
    url, code, msg, hdrs, fp
):

Base class for server exceptions (code 5xx) from url* functions

---

source

urlopener

def urlopener():

Call self as a function.

---

source

urlopen

def urlopen(
    url, data:NoneType=None, headers:NoneType=None, timeout:NoneType=None, **kwargs
):

Like urllib.request.urlopen, but first urlwrap the url, and encode data

With urlopen, the body of the response will also be returned in addition to the message if there is an error:

try: urlopen('https://api.github.com/v3')
except HTTPError as e: 
    print(e.code, e.msg)
    assert 'documentation_url' in e.msg

404 Not Found
====Error Body====
{
  "message": "Not Found",
  "documentation_url": "https://docs.github.com/rest",
  "status": "404"
}

---

source

urlread

def urlread(
    url, data:NoneType=None, headers:NoneType=None, decode:bool=True, return_json:bool=False,
    return_headers:bool=False, timeout:NoneType=None, **kwargs
):

Retrieve url, using data dict or kwargs to POST if present

---

source

urljson

def urljson(
    url, data:NoneType=None, headers:NoneType=None, timeout:NoneType=None
):

Retrieve url and decode json

test_eq(urljson('https://httpbin.org/get')['headers']['User-Agent'], url_default_headers['User-Agent'])

---

source

urlclean

def urlclean(
    url
):

Remove fragment, params, and querystring from url if present

test_eq(urlclean('http://a.com/b?c=1#d'), 'http://a.com/b')

---

source

urlretrieve

def urlretrieve(
    url, filename:NoneType=None, reporthook:NoneType=None, data:NoneType=None, headers:NoneType=None,
    timeout:NoneType=None
):

Same as urllib.request.urlretrieve but also works with Request objects

---

source

urldest

def urldest(
    url, dest:NoneType=None
):

Call self as a function.

---

source

urlsave

def urlsave(
    url, dest:NoneType=None, reporthook:NoneType=None, headers:NoneType=None, timeout:NoneType=None
):

Retrieve url and save based on its name

#skip
with tempfile.TemporaryDirectory() as d: urlsave('http://www.google.com/index.html', d)

---

source

urlvalid

def urlvalid(
    x
):

Test if x is a valid URL

assert urlvalid('http://www.google.com/')
assert not urlvalid('www.google.com/')
assert not urlvalid(1)

Basic client/server

---

source

start_server

def start_server(
    port, host:NoneType=None, dgram:bool=False, reuse_addr:bool=True, n_queue:NoneType=None
):

Create a socket server on port, with optional host, of type dgram

You can create a TCP client and server pass an int as port and optional host. host defaults to your main network interface if not provided. You can create a Unix socket client and server by passing a string to port. A SOCK_STREAM socket is created by default, unless you pass dgram=True, in which case a SOCK_DGRAM socket is created. n_queue sets the listening queue size.

---

source

start_client

def start_client(
    port, host:NoneType=None, dgram:bool=False
):

Create a socket client on port, with optional host, of type dgram

---

source

tobytes

def tobytes(
    s:str
)->bytes:

Convert s into HTTP-ready bytes format

test_eq(tobytes('foo\nbar'), b'foo\r\nbar')

---

source

http_response

def http_response(
    body:NoneType=None, status:int=200, hdrs:NoneType=None, **kwargs
):

Create an HTTP-ready response, adding kwargs to hdrs

exp = b'HTTP/1.1 200 OK\r\nUser-Agent: me\r\nContent-Length: 4\r\n\r\nbody'
test_eq(http_response('body', 200, User_Agent='me'), exp)

---

source

recv_once

def recv_once(
    host:str='localhost', port:int=8000
):

Spawn a thread to receive a single HTTP request and store in d['r']