Skip to content

zimscraperlib.zim.providers

libzim Providers accepting a ref arg to keep it away from garbage collection

Use case is to pass it the Item instance that created the Provider so that the Item lives longer than the provider, thus allowing: - to keep a single copy of the data if it is to be indexed (and thus Provider instanced twice) - to release whatever needs to be once we know data won't be fetched anymore

Classes:

FileLikeProvider 

FileLikeProvider(
    fileobj: BytesIO,
    size: int | None = None,
    ref: object | None = None,
)

Bases: ContentProvider

Provider referrencing a file-like object

Use this to keep a single-copy of a content in memory. Useful for indexed content

Methods:

Attributes:

Source code in src/zimscraperlib/zim/providers.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    fileobj: io.BytesIO,
    size: int | None = None,
    ref: object | None = None,
):
    super().__init__()
    self.ref = ref
    self.fileobj = fileobj
    self.size = size

    if self.size is None:
        self.size = size or self.fileobj.seek(0, io.SEEK_END)
        self.fileobj.seek(0, io.SEEK_SET)

fileobj instance-attribute 

fileobj = fileobj

ref instance-attribute 

ref = ref

size instance-attribute 

size = size

gen_blob 

gen_blob() -> Generator[Blob]
Source code in src/zimscraperlib/zim/providers.py 
60
61
def gen_blob(self) -> Generator[libzim.writer.Blob]:
    yield libzim.writer.Blob(self.fileobj.getvalue())  # pragma: no cover

get_size 

get_size() -> int
Source code in src/zimscraperlib/zim/providers.py 
57
58
def get_size(self) -> int:
    return getattr(self, "size", -1)

FileProvider 

FileProvider(
    filepath: Path,
    size: int | None = None,
    ref: object | None = None,
)

Bases: FileProvider

Attributes:

Source code in src/zimscraperlib/zim/providers.py 
20
21
22
23
24
25
26
27
def __init__(
    self,
    filepath: pathlib.Path,
    size: int | None = None,  # noqa: ARG002
    ref: object | None = None,
):
    super().__init__(filepath)
    self.ref = ref

ref instance-attribute 

ref = ref

StringProvider 

StringProvider(
    content: str | bytes, ref: object | None = None
)

Bases: StringProvider

Attributes:

Source code in src/zimscraperlib/zim/providers.py 
31
32
33
def __init__(self, content: str | bytes, ref: object | None = None):
    super().__init__(content)
    self.ref = ref

ref instance-attribute 

ref = ref

URLProvider 

URLProvider(
    url: str,
    size: int | None = None,
    ref: object | None = None,
)

Bases: ContentProvider

Provider downloading content as it is consumed by the libzim

Useful for non-indexed content for which feed() is called only once

Methods:

Attributes:

Source code in src/zimscraperlib/zim/providers.py 
69
70
71
72
73
74
75
76
77
78
def __init__(self, url: str, size: int | None = None, ref: object | None = None):
    super().__init__()
    self.url = url
    self.size = size if size is not None else self.get_size_of(url)
    self.ref = ref

    session = requests.Session()
    session.mount("http", get_retry_adapter())
    self.resp = session.get(url, stream=True)
    self.resp.raise_for_status()

ref instance-attribute 

ref = ref

resp instance-attribute 

resp = session.get(url, stream=True)

size instance-attribute 

size = size if size is not None else self.get_size_of(url)

url instance-attribute 

url = url

gen_blob 

gen_blob() -> Generator[Blob]
Source code in src/zimscraperlib/zim/providers.py 
91
92
93
94
95
def gen_blob(self) -> Generator[libzim.writer.Blob]:  # pragma: no cover
    for chunk in self.resp.iter_content(10 * 1024):
        if chunk:
            yield libzim.writer.Blob(chunk)
    yield libzim.writer.Blob(b"")

get_size 

get_size() -> int
Source code in src/zimscraperlib/zim/providers.py 
88
89
def get_size(self) -> int:
    return getattr(self, "size", -1)

get_size_of staticmethod 

get_size_of(url: str) -> int | None
Source code in src/zimscraperlib/zim/providers.py 
80
81
82
83
84
85
86
@staticmethod
def get_size_of(url: str) -> int | None:
    _, headers = stream_file(url, byte_stream=io.BytesIO(), only_first_block=True)
    try:
        return int(headers["Content-Length"])
    except Exception:
        return None