Skip to content

Instantly share code, notes, and snippets.

@debedb

debedb/WrappedStreamingBody

Last active July 1, 2025 17:33
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save debedb/2e5cbeb54e43f031eaf0 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save debedb/2e5cbeb54e43f031eaf0 to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. debedb revised this gist Mar 13, 2016. 1 changed file with 1 addition and 1 deletion.
    2 changes: 1 addition & 1 deletion WrappedStreamingBody
    View file
    Original file line number Diff line number Diff line change
    @@ -63,7 +63,7 @@ class WrappedStreamingBody:
    # print("Calling %s" % attr)
    if attr == 'tell':
    return self.tell
    elif attr == 'seek':
    elif attr == 'seek':
    return self.seek
    elif attr == 'read':
    return self.read
  2. debedb revised this gist Mar 13, 2016. 1 changed file with 37 additions and 15 deletions.
    52 changes: 37 additions & 15 deletions WrappedStreamingBody
    View file
    Original file line number Diff line number Diff line change
    @@ -1,29 +1,45 @@
    class WrappedStreamingBody:
    """
    Wrap boto3's StreamingBody object to provide enough fileobj functionality so that GzipFile is
    satisfied, which is useful for processing files from S3 in AWS Lambda which have been gzipped.
    Fairly common use case.

    Sometimes duck typing is awesome.

    """
    Wrap boto3's StreamingBody object to provide enough
    fileobj functionality so that GzipFile is
    satisfied. Sometimes duck typing is awesome.

    https://gist.github.com/debedb/2e5cbeb54e43f031eaf0
    TODO that gist does not have the EOF fix!

    """
    def __init__(self, sb, size):
    # The StreamingBody we're wrapping
    self.sb = sb
    # Initial position
    self.pos = 0
    # Size of the object
    self.size = size

    def tell(self):
    # print("In tell(): %s" % self.pos)
    #print("In tell()")
    return self.pos

    def readline(self):
    #print("Calling readline()")
    try:
    retval = self.sb.readline()
    except struct.error:
    raise EOFError()
    self.pos += len(retval)
    return retval


    def read(self, n=None):
    # print("In read(%s)" % n)
    retval = self.sb.read(n)
    if retval == "":
    raise EOFError()
    self.pos += len(retval)
    return retval

    def seek(self, offset, whence=0):
    retval = self.pos
    #print("Calling seek()")
    retval = self.pos
    if whence == 2:
    if offset == 0:
    retval = self.size
    @@ -36,18 +52,24 @@ class WrappedStreamingBody:
    retval = self.size
    else:
    retval = offset
    # print("In seek(%s, %s): %s, size is %s" % (offset, whence, retval, self.size))
    # print("In seek(%s, %s): %s, size is %s" % (offset, whence, retval, self.size))
    self.pos = retval
    return retval

    def __str__(self):
    return "WrappedBody"

    def __getattr__(self, attr):
    print("Calling %s" % attr)
    # print("Calling %s" % attr)
    if attr == 'tell':
    return self.tell
    elif attr == 'seek':
    elif attr == 'seek':
    return self.seek
    elif attr == 'read':
    return self.read
    elif attr == 'readline':
    return self.readline
    elif attr == '__str__':
    return self.__str__
    else:
    return getattr(self.sb, attr)

    return getattr(self.sb, attr)
  3. debedb revised this gist Mar 7, 2016. No changes.
  4. debedb created this gist Mar 7, 2016.
    53 changes: 53 additions & 0 deletions WrappedStreamingBody
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,53 @@
    class WrappedStreamingBody:
    """
    Wrap boto3's StreamingBody object to provide enough fileobj functionality so that GzipFile is
    satisfied, which is useful for processing files from S3 in AWS Lambda which have been gzipped.
    Fairly common use case.

    Sometimes duck typing is awesome.

    """
    def __init__(self, sb, size):
    self.sb = sb
    self.pos = 0
    self.size = size

    def tell(self):
    # print("In tell(): %s" % self.pos)
    return self.pos

    def read(self, n=None):
    # print("In read(%s)" % n)
    retval = self.sb.read(n)
    self.pos += len(retval)
    return retval

    def seek(self, offset, whence=0):
    retval = self.pos
    if whence == 2:
    if offset == 0:
    retval = self.size
    else:
    raise Exception("Unsupported")
    else:
    if whence == 1:
    offset = self.pos + offset
    if offset > self.size:
    retval = self.size
    else:
    retval = offset
    # print("In seek(%s, %s): %s, size is %s" % (offset, whence, retval, self.size))
    self.pos = retval
    return retval

    def __getattr__(self, attr):
    print("Calling %s" % attr)
    if attr == 'tell':
    return self.tell
    elif attr == 'seek':
    return self.seek
    elif attr == 'read':
    return self.read
    else:
    return getattr(self.sb, attr)