add aiohttp.MozillaCookieJar to load/save cookies.txt files

[milahu]

Is your feature request related to a problem?

aiohttp.CookieJar can only load/save cookies from/to cookies.pickle files
but not from/to cookies.txt files

Describe the solution you'd like

there should be aiohttp.MozillaCookieJar

Describe alternatives you've considered

write my own class AiohttpMozillaCookieJar(aiohttp.CookieJar)

AiohttpMozillaCookieJar.py

# NOTE this is not-really tested... its just a draft

from http.cookiejar import MozillaCookieJar
from http.cookies import BaseCookie, Morsel, SimpleCookie
from collections import defaultdict
from pathlib import Path
from datetime import datetime
from http.cookiejar import Cookie
from typing import Union


from aiohttp import CookieJar


PathLike = Union[str, "os.PathLike[str]"]


class AiohttpMozillaCookieJar(CookieJar):
    """
    load/save cookies from/to a cookies.txt file with aiohttp

    convert between http.cookiejar.MozillaCookieJar and aiohttp.CookieJar

    import aiohttp
    from AiohttpMozillaCookieJar import AiohttpMozillaCookieJar
    jar = AiohttpMozillaCookieJar()
    cookies_txt_path = "cookies.txt"
    jar.load(cookies_txt_path)
    async with aiohttp.ClientSession(cookie_jar=jar) as session:
        url = "..."
        response = await aiohttp_session.get(url)
    jar.save(cookies_txt_path)

    author: Milan Hauth <milahu@gmail.com>
    license: MIT License
    """

    def load(self, file_path):

        file_path = Path(file_path)

        jar_1 = MozillaCookieJar()
        jar_1.load(file_path)

        # Cookie in jar_1 -> Morsel in jar_2

        # jar_1._cookies # dict
        # jar_1._cookies[domain] # dict
        # jar_1._cookies[domain][path] # dict
        # jar_1._cookies[domain][path][name] # Cookie

        # jar_2._cookies # collections.defaultdict(SimpleCookie)
        # jar_2._cookies[(domain, path)] # SimpleCookie
        # jar_2._cookies[(domain, path)][name] # Morsel

        for cookie_1 in jar_1:

            morsel_2 = Morsel()

            domain = cookie_1.domain
            path = cookie_1.path
            name = cookie_1.name

            if name.lower() in Morsel._reserved:
                #print(f"illegal morsel name: {name}")
                continue

            for key in (
                'path',
                'comment',
                'domain',
                'secure',
                'version',
            ):
                if (value := getattr(cookie_1, key)) is not None:
                    morsel_2[key] = value

            morsel_2._key = cookie_1.name
            morsel_2._value = cookie_1.value
            morsel_2._coded_value = cookie_1.value

            try:
                morsel_2['expires'] = datetime.fromtimestamp(cookie_1.expires).strftime(
                    "%a, %d %b %Y %H:%M:%S GMT"
                )
            except (
                OSError,  # Invalid argument (value of cookie.expires is invalid)
                TypeError,  # cookie.expires is None
            ):
                pass

            for key in ('HttpOnly', 'SameSite'):
                if (value := cookie_1.get_nonstandard_attr(key, None)) is not None:
                    morsel_2[key] = value

            self._cookies[(domain, path)][name] = morsel_2


    def save(self, file_path: PathLike) -> None:

        file_path = Path(file_path)

        jar_1 = MozillaCookieJar()

        # Morsel in jar_2 -> Cookie in jar_1

        # jar_2._cookies # collections.defaultdict(SimpleCookie)
        # jar_2._cookies[(domain, path)] # SimpleCookie
        # jar_2._cookies[(domain, path)][name] # Morsel

        # jar_1._cookies # dict
        # jar_1._cookies[domain] # dict
        # jar_1._cookies[domain][path] # dict
        # jar_1._cookies[domain][path][name] # Cookie

        for (domain, path), cookie_2 in self._cookies.items():

            for name, morsel_2 in cookie_2.items():

                try:
                    expires = self._expirations[(domain, path, name)].timestamp()
                except KeyError:
                    try:
                        expires = datetime.strptime(morsel_2["expires"], "%a, %d %b %Y %H:%M:%S %Z").timestamp()
                    except ValueError:
                        # morsel_2["expires"] can be empty string
                        expires = None

                cookie_1 = Cookie(
                    0, # version
                    name, # name
                    morsel_2.value, # value
                    None, # port
                    False, # port_specified
                    domain, # domain
                    False, # domain_specified
                    False, # domain_initial_dot
                    path, # path
                    False, # path_specified
                    morsel_2["secure"], # secure
                    expires, # expires
                    False, # discard
                    None, # comment
                    None, # comment_url
                    {}, # rest
                )
                if not domain in jar_1._cookies:
                    jar_1._cookies[domain] = dict()
                if not path in jar_1._cookies[domain]:
                    jar_1._cookies[domain][path] = dict()
                jar_1._cookies[domain][path][name] = cookie_1

        jar_1.save(file_path)

test-aiohttp-cookies.py

#!/usr/bin/env python

import sys
import asyncio
import aiohttp
from AiohttpMozillaCookieJar import AiohttpMozillaCookieJar
import json
import os

cookie_jar_backend = "pickle"
cookie_jar_backend = "txt"

cookie_jar_path = f"test-aiohttp-cookies.cookies.{cookie_jar_backend}"

cookie_jar_kwargs = dict()
#cookie_jar_kwargs["unsafe"] = True

cookie_jar = None
if cookie_jar_backend == "txt":
    cookie_jar = AiohttpMozillaCookieJar(**cookie_jar_kwargs)
elif cookie_jar_backend == "pickle":
    cookie_jar = aiohttp.CookieJar(**cookie_jar_kwargs)
else:
    raise ValueError(f"cookie_jar_backend {cookie_jar_backend}")

def dump_cookie_jar(cookie_jar):

    for k1 in cookie_jar._cookies:
        domain, path = k1

        #print(f"cookie_jar dir", dir(cookie_jar))

        #print(f"cookie_jar _expirations", cookie_jar._expirations)
        #print(f"cookie_jar _host_only_cookies", cookie_jar._host_only_cookies)
        #print(f"cookie_jar _max_time", cookie_jar._max_time)
        #print(f"cookie_jar _unsafe", cookie_jar._unsafe)

        #print(f"cookie_jar._cookies: domain={domain} path={path}", cookie_jar._cookies[k1])
        #print(f"cookie_jar._cookies: domain={domain} path={path} dir", dir(cookie_jar._cookies[k1]))
        #print(f"cookie_jar._cookies: domain={domain} path={path} items", cookie_jar._cookies[k1].items())
        for name, morsel in cookie_jar._cookies[k1].items():
            print(f"cookie_jar._cookies: domain={domain} path={path} {name}={morsel.value}")

if False:
    # only dump cookiejar
    cookie_jar.load(cookie_jar_path)
    dump_cookie_jar(cookie_jar)
    sys.exit()

async def main():

    session_kwargs = dict()
    # not used?
    session_kwargs["cookie_jar"] = cookie_jar
    session_kwargs["headers"] = dict(session_header="1")

    get_kwargs = dict()
    get_kwargs["headers"] = dict(request_header="1")
    # TypeError: ClientSession._request() got an unexpected keyword argument 'cookie_jar'
    #get_kwargs["cookie_jar"] = cookie_jar

    if os.path.exists(cookie_jar_path):
        print(f"loading cookies from {cookie_jar_path}")
        cookie_jar.load(cookie_jar_path)

    async with aiohttp.ClientSession(**session_kwargs) as aiohttp_session:

        # debug cookies
        # FIXME aiohttp does not send cookies?!

        print(f"aiohttp_session.cookie_jar._cookies:")
        dump_cookie_jar(aiohttp_session.cookie_jar)

        #print(f"aiohttp_session.cookie_jar._cookies httpbin.dev /cookies:\n" + str(aiohttp_session.cookie_jar._cookies[("httpbin.dev", "/cookies")]))

        url_base = "http://httpbin.dev"
        #url_base = "https://httpbin.dev"

        # debug: set cookies
        #url = f"{url_base}/cookies/set?foo=bar"
        url = f"{url_base}/cookies/set?foo=bar&a=1&b=2"
        response = await aiohttp_session.get(url, **get_kwargs)

        # Cookie headers are not here
        #print(f"response.request_info.headers: {response.request_info.headers}")

        response_text = (await response.content.read()).decode("utf8")
        #print(f"response_text: {response_text}")

        # Set-Cookie headers are not here
        #print(f"response.headers: {response.headers}")

        print(f"aiohttp_session.cookie_jar._cookies:")
        dump_cookie_jar(aiohttp_session.cookie_jar)

        #print(f"aiohttp_session.cookie_jar._cookies httpbin.dev /cookies:\n" + str(aiohttp_session.cookie_jar._cookies[("httpbin.dev", "/cookies")]))


        # NOTE cookies are not returned by /headers
        url = f"{url_base}/headers"
        response = await aiohttp_session.get(url, **get_kwargs)

        print(f"response.request_info.headers.Cookie: {response.request_info.headers.get('Cookie')}")
        #print(f"response.request_info dir: {dir(response.request_info)}")

        response_status = response.status
        # TODO debug request headers
        #print(f"response dir: {dir(response)}")
        #print(f"response_status: {response_status}")

        # Set-Cookie headers are not here
        #print(f"headers: {response.headers}")

        response_content = response.content
        response_text = (await response_content.read()).decode("utf8")
        print(f"sent headers: {response_text}")



        # debug: get cookies
        url = f"{url_base}/cookies"
        response = await aiohttp_session.get(url, **get_kwargs)

        print(f"response.request_info.headers.Cookie: {response.request_info.headers.get('Cookie')}")
        #print(f"response.request_info dir: {dir(response.request_info)}")

        response_status = response.status
        # TODO debug request headers
        #print(f"response dir: {dir(response)}")
        #print(f"response_status: {response_status}")

        # Set-Cookie headers are not here
        #print(f"headers: {response.headers}")

        response_content = response.content
        response_text = (await response_content.read()).decode("utf8")
        print(f"sent cookies: {response_text}")

        print(f"aiohttp_session.cookie_jar._cookies:")
        dump_cookie_jar(aiohttp_session.cookie_jar)

        #print(f"aiohttp_session.cookie_jar._cookies httpbin.dev /cookies:\n" + str(aiohttp_session.cookie_jar._cookies[("httpbin.dev", "/cookies")]))

        print(f"saving cookies to {cookie_jar_path}")
        aiohttp_session.cookie_jar.save(cookie_jar_path)

asyncio.get_event_loop().run_until_complete(main())

Related component

Client

Additional context

stackoverflow:

How to use cookies exported from chrome in python aiohttp?

I'm trying to visit some website using aiohttp/python. Currently I can export the site's cookies with extensions in Google Chrome, formatted like

.domain.name TRUE / FALSE 1547016401 cookies values

Code of Conduct

• I agree to follow the aio-libs Code of Conduct