API References

curl low levl APIs

Curl

class curl_cffi.Curl(cacert: str = '', debug: bool = False, handle=None)[source]

Wrapper for curl_easy_* functions of libcurl.

__init__(cacert: str = '', debug: bool = False, handle=None) → None[source]
Parameters:

  • cacert – CA cert path to use, by default, curl_cffi uses certs from certifi.

  • debug – whether to show curl debug messages.

  • handle – a curl handle instance from curl_easy_init.

debug() → None[source]

Set debug to True

setopt(option: CurlOpt, value: Any) → int[source]

Wrapper for curl_easy_setopt.

Parameters:

  • option – option to set, using constants from CurlOpt enum

  • value – value to set, strings will be handled automatically

Returns:

0 if no error, see CurlECode.

getinfo(option: CurlInfo) → bytes | int | float | List[source]

Wrapper for curl_easy_getinfo. Gets information in response after curl perform.

Parameters:

option – option to get info of, using constants from CurlInfo enum

Returns:

value retrieved from last perform.

version() → bytes[source]

Get the underlying libcurl version.

impersonate(target: str, default_headers: bool = True) → int[source]

Set the browser type to impersonate.

Parameters:

  • target – browser to impersonate.

  • default_headers – whether to add default headers, like User-Agent.

Returns:

0 if no error.

perform(clear_headers: bool = True) → None[source]

Wrapper for curl_easy_perform, performs a curl request.

Parameters:

clear_headers – clear header slist used in this perform

Raises:

CurlError – if the perform was not successful.

duphandle() → Curl[source]

Wrapper for curl_easy_duphandle.

This is not a full copy of entire curl object in python. For example, headers handle is not copied, you have to set them again.

reset() → None[source]

Reset all curl options, wrapper for curl_easy_reset.

Extract cookies.SimpleCookie from header lines.

Parameters:

headers – list of headers in bytes.

Returns:

A parsed cookies.SimpleCookie instance.

static get_reason_phrase(status_line: bytes) → bytes[source]

Extract reason phrase, like OK, Not Found from response status line.

static parse_status_line(status_line: bytes) → Tuple[CurlHttpVersion, int, bytes][source]

Parse status line.

Returns:

http_version, status_code, and reason phrase

close() → None[source]

Close and cleanup curl handle, wrapper for curl_easy_cleanup.

ws_recv(n: int = 1024) → Tuple[bytes, Any][source]

Receive a frame from a websocket connection.

Parameters:

n – maximum data to receive.

Returns:

a tuple of frame content and curl frame meta struct.

Raises:

CurlError – if failed.

ws_send(payload: bytes, flags: CurlWsFlag = CurlWsFlag.BINARY) → int[source]

Send data to a websocket connection.

Parameters:

  • payload – content to send.

  • flags – websocket flag to set for the frame, default: binary.

Returns:

0 if no error.

Raises:

CurlError – if failed.

ws_close() → None[source]

Send the close frame.

AsyncCurl

class curl_cffi.AsyncCurl(cacert: str = '', loop=None)[source]

Wrapper around curl_multi handle to provide asyncio support. It uses the libcurl socket_action APIs.

__init__(cacert: str = '', loop=None)[source]
Parameters:

  • cacert – CA cert path to use, by default, curl_cffi uses certs from certifi.

  • loop – EventLoop to use.

add_handle(curl: Curl)[source]

Add a curl handle to be managed by curl_multi. This is the equivalent of perform in the async world.

remove_handle(curl: Curl)[source]

Cancel a future for given curl handle.

set_result(curl: Curl)[source]

Mark a future as done for given curl handle.

set_exception(curl: Curl, exception)[source]

Raise exception of a future for given curl handle.

setopt(option, value)[source]

Wrapper around curl_multi_setopt.

socket_action(sockfd: int, ev_bitmask: int) → int[source]

Call libcurl socket_action function

process_data(sockfd: int, ev_bitmask: int)[source]

Call curl_multi_info_read to read data for given socket.

async close()[source]

Close and cleanup running timers, readers, writers and handles.

CurlMime

class curl_cffi.CurlMime(curl: Curl | None = None)[source]

Wrapper for the curl_mime_ API.

__init__(curl: Curl | None = None)[source]
Parameters:

curl – Curl instance to use.

addpart(name: str, *, content_type: str | None = None, filename: str | None = None, local_path: str | bytes | Path | None = None, data: bytes | None = None) → None[source]

Add a mime part for a mutlipart html form.

Note: You can only use either local_path or data, not both.

Parameters:

  • name – name of the field.

  • content_type – content_type for the field. for example: image/png.

  • filename – filename for the server.

  • local_path – file to upload on local disk.

  • data – file content to upload.

classmethod from_list(files: List[dict])[source]

Create a multipart instance from a list of dict, for keys, see addpart

attach(curl: Curl | None = None) → None[source]

Attach the mime instance to a curl instance.

close() → None[source]

Close the mime instance and underlying files. This method must be called after perform or request.

Constants

Enum values used by setopt and getinfo can be accessed from CurlOpt and CurlInfo.

class curl_cffi.CurlOpt(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

CULROPT_ constancs extracted from libcurl, see: https://curl.se/libcurl/c/curl_easy_setopt.html

class curl_cffi.CurlInfo(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

CURLINFO_ constancs extracted from libcurl, see: https://curl.se/libcurl/c/curl_easy_getinfo.html

class curl_cffi.CurlMOpt(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

CURLMOPT_ constancs extracted from libcurl, see: https://curl.se/libcurl/c/curl_multi_setopt.html

class curl_cffi.CurlECode(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

CURLECODE_ constancs extracted from libcurl, see: https://curl.se/libcurl/c/libcurl-errors.html

class curl_cffi.CurlHttpVersion(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

CURL_HTTP_VERSION constants extracted from libcurl, see comments for details

class curl_cffi.CurlWsFlag(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

CURL_WS_FLAG constancs extracted from libcurl, see comments for details

requests API

request method

requests.get, requests.post, etc are just aliases of .request(METHOD, ...)

curl_cffi.requests.request(method: str, url: str, params: Dict | List | Tuple | None = None, data: Dict[str, str] | List[Tuple] | str | BytesIO | bytes | None = None, json: dict | None = None, headers: Headers | Mapping[str, str | None] | Mapping[bytes, bytes | None] | Sequence[Tuple[str, str]] | Sequence[Tuple[bytes, bytes]] | Sequence[str | bytes] | None = None, cookies: Cookies | CookieJar | Dict[str, str] | List[Tuple[str, str]] | None = None, files: Dict | None = None, auth: Tuple[str, str] | None = None, timeout: float | Tuple[float, float] = 30, allow_redirects: bool = True, max_redirects: int = -1, proxies: Dict[str, str] | None = None, proxy: str | None = None, proxy_auth: Tuple[str, str] | None = None, verify: bool | None = None, referer: str | None = None, accept_encoding: str | None = 'gzip, deflate, br', content_callback: Callable | None = None, impersonate: str | BrowserType | None = None, thread: Literal['eventlet', 'gevent'] | None = None, default_headers: bool | None = None, default_encoding: str | Callable[[bytes], str] = 'utf-8', curl_options: dict | None = None, http_version: CurlHttpVersion | None = None, debug: bool = False, interface: str | None = None, cert: str | Tuple[str, str] | None = None, stream: bool = False, max_recv_speed: int = 0, multipart: CurlMime | None = None) → Response[source]

Send an http request.

Parameters:

  • method – http method for the request: GET/POST/PUT/DELETE etc.

  • url – url for the requests.

  • params – query string for the requests.

  • data – form values or binary data to use in body, Content-Type: application/x-www-form-urlencoded will be added if a dict is given.

  • json – json values to use in body, Content-Type: application/json will be added automatically.

  • headers – headers to send.

  • cookies – cookies to use.

  • files – not supported, use multipart instead.

  • auth – HTTP basic auth, a tuple of (username, password), only basic auth is supported.

  • timeout – how many seconds to wait before giving up.

  • allow_redirects – whether to allow redirection.

  • max_redirects – max redirect counts, default unlimited(-1).

  • proxies – dict of proxies to use, format: {"http": proxy_url, "https": proxy_url}.

  • proxy – proxy to use, format: “http://user@pass:proxy_url”. Can’t be used with proxy parameter.

  • proxy_auth – HTTP basic auth for proxy, a tuple of (username, password).

  • verify – whether to verify https certs.

  • referer – shortcut for setting referer header.

  • accept_encoding – shortcut for setting accept-encoding header.

  • content_callback – a callback function to receive response body. def callback(chunk: bytes) -> None:

  • impersonate – which browser version to impersonate.

  • thread – work with other thread implementations. choices: eventlet, gevent.

  • default_headers – whether to set default browser headers.

  • default_encoding – encoding for decoding response content if charset is not found in headers. Defaults to “utf-8”. Can be set to a callable for automatic detection.

  • curl_options – extra curl options to use.

  • http_version – limiting http version, http2 will be tries by default.

  • debug – print extra curl debug info.

  • interface – which interface use in request to server.

  • multipart – upload files using the multipart format, see.

Returns:

A Response object.

Sessions

class curl_cffi.requests.Session(curl: Curl | None = None, thread: Literal['eventlet', 'gevent'] | None = None, use_thread_local_curl: bool = True, **kwargs)[source]

A request session, cookies and connections will be reused. This object is thread-safe, but it’s recommended to use a seperate session for each thread.

__init__(curl: Curl | None = None, thread: Literal['eventlet', 'gevent'] | None = None, use_thread_local_curl: bool = True, **kwargs)[source]

Parameters set in the init method will be override by the same parameter in request method.

Parameters:

  • curl – curl object to use in the session. If not provided, a new one will be created. Also, a fresh curl object will always be created when accessed from another thread.

  • thread – thread engine to use for working with other thread implementations. choices: eventlet, gevent., possible values: eventlet, gevent.

  • headers – headers to use in the session.

  • cookies – cookies to add in the session.

  • auth – HTTP basic auth, a tuple of (username, password), only basic auth is supported.

  • proxies – dict of proxies to use, format: {“http”: proxy_url, “https”: proxy_url}.

  • proxy – proxy to use, format: “http://proxy_url”. Cannot be used with the above parameter.

  • proxy_auth – HTTP basic auth for proxy, a tuple of (username, password).

  • base_url – absolute url to use for relative urls.

  • params – query string for the session.

  • verify – whether to verify https certs.

  • timeout – how many seconds to wait before giving up.

  • trust_env – use http_proxy/https_proxy and other environments, default True.

  • allow_redirects – whether to allow redirection.

  • max_redirects – max redirect counts, default unlimited(-1).

  • impersonate – which browser version to impersonate in the session.

  • interface – which interface use in request to server.

  • default_encoding – encoding for decoding response content if charset is not found in headers. Defaults to “utf-8”. Can be set to a callable for automatic detection.

Notes

This class can be used as a context manager.

from curl_cffi.requests import Session

with Session() as s:
    r = s.get("https://example.com")
request(method: str, url: str, params: ~typing.Dict | ~typing.List | ~typing.Tuple | None = None, data: ~typing.Dict[str, str] | ~typing.List[~typing.Tuple] | str | ~_io.BytesIO | bytes | None = None, json: dict | None = None, headers: ~curl_cffi.requests.headers.Headers | ~collections.abc.Mapping[str, str | None] | ~collections.abc.Mapping[bytes, bytes | None] | ~collections.abc.Sequence[~typing.Tuple[str, str]] | ~collections.abc.Sequence[~typing.Tuple[bytes, bytes]] | ~collections.abc.Sequence[str | bytes] | None = None, cookies: ~curl_cffi.requests.cookies.Cookies | ~http.cookiejar.CookieJar | ~typing.Dict[str, str] | ~typing.List[~typing.Tuple[str, str]] | None = None, files: ~typing.Dict | None = None, auth: ~typing.Tuple[str, str] | None = None, timeout: float | ~typing.Tuple[float, float] | object | None = <object object>, allow_redirects: bool | None = None, max_redirects: int | None = None, proxies: ~typing.Dict[str, str] | None = None, proxy: str | None = None, proxy_auth: ~typing.Tuple[str, str] | None = None, verify: bool | None = None, referer: str | None = None, accept_encoding: str | None = 'gzip, deflate, br', content_callback: ~typing.Callable | None = None, impersonate: str | ~curl_cffi.requests.session.BrowserType | None = None, default_headers: bool | None = None, default_encoding: str | ~typing.Callable[[bytes], str] = 'utf-8', http_version: ~curl_cffi.const.CurlHttpVersion | None = None, interface: str | None = None, cert: str | ~typing.Tuple[str, str] | None = None, stream: bool = False, max_recv_speed: int = 0, multipart: ~curl_cffi.curl.CurlMime | None = None) → Response[source]

Send the request, see requests.request for details on parameters.

stream(*args, **kwargs)[source]

Equivalent to with request(..., stream=True) as r:

ws_connect(url, *args, on_message: Callable[[WebSocket, bytes], None] | None = None, on_error: Callable[[WebSocket, CurlError], None] | None = None, on_open: Callable | None = None, on_close: Callable | None = None, **kwargs) → WebSocket[source]

Connects to a websocket url.

Parameters:

  • url – the ws url to connect.

  • on_message – message callback, def on_message(ws, str)

  • on_error – error callback, def on_error(ws, error)

  • on_open – open callback, def on_open(ws)

  • on_cloes – close callback, def on_close(ws)

Other parameters are the same as .request

Returns:

a ws instance to communicate with the server.

class curl_cffi.requests.AsyncSession(*, loop=None, async_curl: AsyncCurl | None = None, max_clients: int = 10, **kwargs)[source]

An async request session, cookies and connections will be reused.

__init__(*, loop=None, async_curl: AsyncCurl | None = None, max_clients: int = 10, **kwargs)[source]

Parameters set in the init method will be override by the same parameter in request method.

Parameters:

  • loop – loop to use, if not provided, the running loop will be used.

  • async_curl – [AsyncCurl](/api/curl_cffi#curl_cffi.AsyncCurl) object to use.

  • max_clients – maxmium curl handle to use in the session, this will affect the concurrency ratio.

  • headers – headers to use in the session.

  • cookies – cookies to add in the session.

  • auth – HTTP basic auth, a tuple of (username, password), only basic auth is supported.

  • proxies – dict of proxies to use, format: {“http”: proxy_url, “https”: proxy_url}.

  • proxy – proxy to use, format: “http://proxy_url”. Cannot be used with the above parameter.

  • proxy_auth – HTTP basic auth for proxy, a tuple of (username, password).

  • base_url – absolute url to use for relative urls.

  • params – query string for the session.

  • verify – whether to verify https certs.

  • timeout – how many seconds to wait before giving up.

  • trust_env – use http_proxy/https_proxy and other environments, default True.

  • allow_redirects – whether to allow redirection.

  • max_redirects – max redirect counts, default unlimited(-1).

  • impersonate – which browser version to impersonate in the session.

  • default_encoding – encoding for decoding response content if charset is not found in headers. Defaults to “utf-8”. Can be set to a callable for automatic detection.

Notes

This class can be used as a context manager, and it’s recommended to use via async with. However, unlike aiohttp, it is not required to use with.

from curl_cffi.requests import AsyncSession

# recommended.
async with AsyncSession() as s:
    r = await s.get("https://example.com")

s = AsyncSession()  # it also works.
async request(method: str, url: str, params: ~typing.Dict | ~typing.List | ~typing.Tuple | None = None, data: ~typing.Dict[str, str] | ~typing.List[~typing.Tuple] | str | ~_io.BytesIO | bytes | None = None, json: dict | None = None, headers: ~curl_cffi.requests.headers.Headers | ~collections.abc.Mapping[str, str | None] | ~collections.abc.Mapping[bytes, bytes | None] | ~collections.abc.Sequence[~typing.Tuple[str, str]] | ~collections.abc.Sequence[~typing.Tuple[bytes, bytes]] | ~collections.abc.Sequence[str | bytes] | None = None, cookies: ~curl_cffi.requests.cookies.Cookies | ~http.cookiejar.CookieJar | ~typing.Dict[str, str] | ~typing.List[~typing.Tuple[str, str]] | None = None, files: ~typing.Dict | None = None, auth: ~typing.Tuple[str, str] | None = None, timeout: float | ~typing.Tuple[float, float] | object | None = <object object>, allow_redirects: bool | None = None, max_redirects: int | None = None, proxies: ~typing.Dict[str, str] | None = None, proxy: str | None = None, proxy_auth: ~typing.Tuple[str, str] | None = None, verify: bool | None = None, referer: str | None = None, accept_encoding: str | None = 'gzip, deflate, br', content_callback: ~typing.Callable | None = None, impersonate: str | ~curl_cffi.requests.session.BrowserType | None = None, default_headers: bool | None = None, default_encoding: str | ~typing.Callable[[bytes], str] = 'utf-8', http_version: ~curl_cffi.const.CurlHttpVersion | None = None, interface: str | None = None, cert: str | ~typing.Tuple[str, str] | None = None, stream: bool = False, max_recv_speed: int = 0, multipart: ~curl_cffi.curl.CurlMime | None = None)[source]

Send the request, see curl_cffi.requests.request for details on parameters.

stream(*args, **kwargs)[source]

Equivalent to async with request(..., stream=True) as r:

async close() → None[source]

Close the session.

async ws_connect(url, *args, **kwargs)[source]

Headers

class curl_cffi.requests.Headers(headers: Headers | Mapping[str, str | None] | Mapping[bytes, bytes | None] | Sequence[Tuple[str, str]] | Sequence[Tuple[bytes, bytes]] | Sequence[str | bytes] | None = None, encoding: str | None = None)[source]

HTTP headers, as a case-insensitive multi-dict.

property encoding: str

Header encoding is mandated as ascii, but we allow fallbacks to utf-8 or iso-8859-1.

raw()

Returns a list of the raw header items, as byte pairs.

multi_items() → List[Tuple[str, str]][source]

Return a list of (key, value) pairs of headers. Allow multiple occurrences of the same key without concatenating into a single comma separated value.

get(key: str, default: Any = None) → Any[source]

Return a header value. If multiple occurrences of the header occur then concatenate them together with commas.

get_list(key: str, split_commas: bool = False) → List[str][source]

Return a list of all header values for a given key. If split_commas=True is passed, then any comma separated header values are split into multiple return strings.

update([E, ]**F) → None.  Update D from mapping/iterable E and F.[source]

If E present and has a .keys() method, does: for k in E: D[k] = E[k] If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v In either case, this is followed by: for k, v in F.items(): D[k] = v

__getitem__(key: str) → str[source]

Return a single header value. If there are multiple headers with the same key, then we concatenate them with commas. See: https://tools.ietf.org/html/rfc7230#section-3.2.2

__setitem__(key: str, value: str) → None[source]

Set the header key to value, removing any duplicate entries. Retains insertion order.

__delitem__(key: str) → None[source]

Remove the header key.

Cookies

class curl_cffi.requests.Cookies(cookies: Cookies | CookieJar | Dict[str, str] | List[Tuple[str, str]] | None = None)[source]

HTTP Cookies, as a mutable mapping.

set(name: str, value: str, domain: str = '', path: str = '/', secure=False) → None[source]

Set a cookie value by name. May optionally include domain and path.

get(name: str, default: str | None = None, domain: str | None = None, path: str | None = None) → str | None[source]

Get a cookie by name. May optionally include domain and path in order to specify exactly which cookie to retrieve.

delete(name: str, domain: str | None = None, path: str | None = None) → None[source]

Delete a cookie by name. May optionally include domain and path in order to specify exactly which cookie to delete.

clear(domain: str | None = None, path: str | None = None) → None[source]

Delete all cookies. Optionally include a domain and path in order to only delete a subset of all the cookies.

update([E, ]**F) → None.  Update D from mapping/iterable E and F.[source]

If E present and has a .keys() method, does: for k in E: D[k] = E[k] If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v In either case, this is followed by: for k, v in F.items(): D[k] = v

__getitem__(name: str) → str[source]
__setitem__(name: str, value: str) → None[source]
__delitem__(name: str) → None[source]

Request, Response and WebSocket

class curl_cffi.requests.Request(url: str, headers: Headers, method: str)[source]

Representing a sent request.

class curl_cffi.requests.Response(curl: Curl | None = None, request: Request | None = None)[source]

Contains information the server sends.

url

url used in the request.

content

response body in bytes.

text

response body in str.

status_code

http status code.

reason

http response reason, such as OK, Not Found.

ok

is status_code in [200, 400)?

headers

response headers.

cookies

response cookies.

elapsed

how many seconds the request cost.

encoding

http body encoding.

charset

alias for encoding.

charset_encoding

encoding specified by the Content-Type header.

default_encoding

encoding for decoding response content if charset is not found in headers. Defaults to “utf-8”. Can be set to a callable for automatic detection.

redirect_count

how many redirects happened.

redirect_url

the final redirected url.

http_version

http version used.

history

history redirections, only headers are available.

raise_for_status()[source]

Raise an error if status code is not in [200, 400)

iter_lines(chunk_size=None, decode_unicode=False, delimiter=None)[source]

iterate streaming content line by line, separated by \n.

Copied from: https://requests.readthedocs.io/en/latest/_modules/requests/models/ which is under the License: Apache 2.0

iter_content(chunk_size=None, decode_unicode=False)[source]

iterate streaming content chunk by chunk in bytes.

json(**kw)[source]

return a parsed json object of the content.

close()[source]

Close the streaming connection, only valid in stream mode.

async aiter_lines(chunk_size=None, decode_unicode=False, delimiter=None)[source]

iterate streaming content line by line, separated by \n.

Copied from: https://requests.readthedocs.io/en/latest/_modules/requests/models/ which is under the License: Apache 2.0

async aiter_content(chunk_size=None, decode_unicode=False)[source]

iterate streaming content chunk by chunk in bytes.

async atext() → str[source]

Return a decoded string.

async acontent() → bytes[source]

wait and read the streaming content in one bytes object.

async aclose()[source]

Close the streaming connection, only valid in stream mode.

class curl_cffi.requests.WebSocket(session, curl, on_message: Callable[[WebSocket, bytes], None] | None = None, on_error: Callable[[WebSocket, CurlError], None] | None = None, on_open: Callable[[WebSocket], None] | None = None, on_close: Callable[[WebSocket, int, str], None] | None = None)[source]
__init__(session, curl, on_message: Callable[[WebSocket, bytes], None] | None = None, on_error: Callable[[WebSocket, CurlError], None] | None = None, on_open: Callable[[WebSocket], None] | None = None, on_close: Callable[[WebSocket, int, str], None] | None = None)[source]
recv() → Tuple[bytes, int][source]

Receive a frame as bytes.

libcurl split frames into fragments, so we have to collect all the chunks for a frame.

send(payload: bytes, flags: CurlWsFlag = CurlWsFlag.BINARY)[source]

Send a data frame

run_forever()[source]

libcurl automatically handles pings and pongs.

ref: https://curl.se/libcurl/c/libcurl-ws.html

close(code: int = WsCloseCode.OK, message: bytes = b'')[source]
async arecv() → Tuple[bytes, int][source]
async asend(payload: bytes, flags: CurlWsFlag = CurlWsFlag.BINARY)[source]