🔍 Code Extractor

Base class representing an item (document or folder) in a reMarkable cloud storage system, providing methods for metadata management, file operations, and synchronization.

Purpose

The Item class serves as a base class for Document and Folder types in a reMarkable cloud storage interface. It handles item metadata, provides async/sync methods for downloading, uploading, and managing items, and includes factory methods for creating instances from metadata or creating new items. The class manages item properties like name, ID, version, parent relationships, and provides file operations with caching support.

Source Code

class Item:

    DOCUMENT = 'DocumentType'
    FOLDER = 'CollectionType'

    @staticmethod
    async def get_by_id(id_):
        return await (await api.get_client()).get_by_id(id_)

    # Decorating a staticmethod is not trivial.  Since this is the only one,
    # we define the _s manually, instead of using add_sync.
    @staticmethod
    def get_by_id_s(id_):
        return trio.run(Item.get_by_id, id_)

    @classmethod
    def from_metadata(cls, metadata):
        type_ = metadata.get('Type')
        if type_ == cls.DOCUMENT:
            return Document(metadata)
        if type_ == cls.FOLDER:
            return Folder(metadata)
        log.error(f"Unknown document type: {type_}")
        return None

    @classmethod
    def new(cls, name, parent_id):
        if issubclass(cls, Document):
            type_ = cls.DOCUMENT
        elif issubclass(cls, Folder):
            type_ = cls.FOLDER
        else:
            log.error(f"Cannot create a new item of class {cls}")
            return None

        metadata = {
            'VissibleName': name,
            'ID': str(uuid.uuid4()),
            'Version': 0,
            'Parent': parent_id,
            'Type': type_
        }
        return cls(metadata)

    def __init__(self, metadata):
        self._metadata = metadata
        self._raw_size = datacache.get_property(self.id, self.version, 'raw_size') or 0
        self._size = datacache.get_property(self.id, self.version, 'size') or 0
        try:
            self._type = FileType[datacache.get_property(self.id, self.version, 'type')]
        except KeyError:
            self._type = None
        self._lock = trio.Lock()

    @property
    def name(self):
        return self._metadata.get('VissibleName')

    @name.setter
    def name(self, value):
        self._metadata['VissibleName'] = value

    @property
    def id(self):
        return self._metadata.get('ID')

    @property
    def version(self):
        return self._metadata.get('Version')

    @property
    def parent(self):
        return self._metadata.get('Parent')

    @parent.setter
    def parent(self, value):
        self._metadata['Parent'] = value

    @property
    def mtime(self):
        return parse_datetime(self._metadata.get('ModifiedClient'))

    @property
    def virtual(self):
        return False

    def __repr__(self):
        return f'<{self.__class__.__name__} "{self.name}">'

    async def _refresh_metadata(self, downloadable=True):
        try:
            self._metadata = await (await api.get_client()).get_metadata(self.id, downloadable)
        except DocumentNotFound:
            log.error(f"Could not update metadata for {self}")

    @add_sync
    @with_lock
    async def download_url(self):
        if not (self._metadata['BlobURLGet'] and
                parse_datetime(self._metadata['BlobURLGetExpires']) > now()):
            await self._refresh_metadata(downloadable=True)
        # This could have failed...
        url = self._metadata['BlobURLGet']
        if url and parse_datetime(self._metadata['BlobURLGetExpires']) > now():
            return url
        return None

    @add_sync
    @with_lock
    async def raw(self):
        if await self.download_url():
            contents_blob = await (await api.get_client()).get_blob(await self.download_url())
            return io.BytesIO(contents_blob)
        return None

    @add_sync
    @with_lock
    async def raw_size(self):
        if not self._raw_size and await self.download_url():
            self._raw_size = await (await api.get_client()).get_blob_size(await self.download_url())
            datacache.set_property(self.id, self.version, 'raw_size', self._raw_size)
        return self._raw_size

    @with_lock
    async def _get_details(self):
        if not self._type and await self.download_url():
            log.debug(f"Getting details for {self}")
            self._type, self._size = await (await api.get_client()).get_file_details(await self.download_url())
            if self._size is None:
                self._size = await self.raw_size()
            datacache.set_property(self.id, self.version, 'size', self._size)
            if self._type != FileType.unknown:
                # Try again the next time we start up.
                datacache.set_property(self.id, self.version, 'type', str(self._type))
            log.debug(f"Details for {self}: type {self._type}, size {self._size}")

    @add_sync
    async def type(self):
        await self._get_details()
        return self._type

    @add_sync
    async def size(self):
        await self._get_details()
        return self._size

    @add_sync
    @with_lock
    async def update_metadata(self):
        if self.virtual:
            raise VirtualItemError('Cannot update virtual items')
        await (await api.get_client()).update_metadata(self)

    @add_sync
    @with_lock
    async def delete(self):
        if self.virtual:
            raise VirtualItemError('Cannot delete virtual items')

        client = await api.get_client()
        folder = self
        while folder.parent:
            folder = await client.get_by_id(folder.parent)
            if folder.id == TRASH_ID:
                return await client.delete(self)
        self.parent = TRASH_ID
        await client.update_metadata(self)

    @add_sync
    @with_lock
    async def upload_raw(self, new_contents):
        if self.virtual:
            raise VirtualItemError('Cannot update virtual items')
        await (await api.get_client()).upload(self, new_contents)

Parameters

Parameter Details

metadata: A dictionary containing item metadata with keys like 'VissibleName' (item name), 'ID' (unique identifier), 'Version' (version number), 'Parent' (parent folder ID), 'Type' (either 'DocumentType' or 'CollectionType'), 'ModifiedClient' (modification timestamp), 'BlobURLGet' (download URL), and 'BlobURLGetExpires' (URL expiration timestamp). This metadata structure matches the reMarkable cloud API format.

Return Value

The __init__ method returns an Item instance. Key method returns include: get_by_id returns an Item instance fetched from the API; from_metadata returns a Document or Folder instance based on metadata type; new returns a new Item instance with generated UUID; download_url returns a string URL or None; raw returns a BytesIO object or None; raw_size and size return integers; type returns a FileType enum value; update_metadata and delete return None but perform side effects.

Class Interface

Methods

Attributes

Dependencies

• trio
• uuid
• io
• logging
• functools
• json
• zipfile

Required Imports

import trio
import uuid
import io
import logging
from const import ROOT_ID
from const import TRASH_ID
from const import FileType
from sync import add_sync
from utils import now
from utils import parse_datetime
from exceptions import DocumentNotFound
from exceptions import VirtualItemError

Conditional/Optional Imports

These imports are only needed under specific conditions:

from  import api

from  import datacache

from rmrl import render

from rmrl import sources

Usage Example

# Async usage
import trio
from item import Item

async def main():
    # Get an existing item by ID
    item = await Item.get_by_id('some-uuid-here')
    print(f"Item name: {item.name}")
    print(f"Item ID: {item.id}")
    
    # Get item details
    file_type = await item.type()
    file_size = await item.size()
    
    # Download item content
    content = await item.raw()
    if content:
        data = content.read()
    
    # Update item name
    item.name = 'New Name'
    await item.update_metadata()
    
    # Delete item (moves to trash)
    await item.delete()

trio.run(main)

# Synchronous usage
item = Item.get_by_id_s('some-uuid-here')
print(item.name)
file_type = item.type_s()
file_size = item.size_s()
content = item.raw_s()

# Create from metadata
metadata = {
    'VissibleName': 'My Document',
    'ID': 'existing-uuid',
    'Version': 1,
    'Parent': 'parent-uuid',
    'Type': 'DocumentType'
}
item = Item.from_metadata(metadata)

Best Practices

• Always use async methods with await in async contexts, or use the _s suffixed synchronous versions
• The class uses a lock (_lock) internally to prevent concurrent modifications - methods decorated with @with_lock are thread-safe
• Metadata is cached and refreshed automatically when needed (e.g., when download URLs expire)
• Use from_metadata() factory method to create proper subclass instances (Document or Folder) from metadata
• Use new() class method to create new items with proper initialization
• Virtual items (where virtual property returns True) cannot be updated or deleted - check before operations
• Delete operations move items to trash rather than permanently deleting them
• Download URLs expire and are automatically refreshed when needed
• File type and size are lazily loaded and cached - first access may be slower
• The class expects Document and Folder subclasses to exist for proper factory method operation
• Always check return values from download_url() and raw() as they may return None on failure
• The @add_sync decorator automatically creates synchronous versions of async methods with _s suffix

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class Document_v1 69.2% similar

  Document class represents a reMarkable document file, extending the Item class to provide document-specific operations like content extraction, uploading, and rendering with annotations.

  From: /tf/active/vicechatdev/rmcl/items.py

• class RemarkableNode 66.5% similar

  A dataclass representing a node (file or folder) in the reMarkable cloud storage system, containing metadata, hierarchy information, and component hashes for documents.

  From: /tf/active/vicechatdev/e-ink-llm/cloudtest/discovery.py

• class Folder 65.2% similar

  Represents a folder item in a file system hierarchy, extending the Item base class with the ability to contain children and be uploaded as a ZIP archive.

  From: /tf/active/vicechatdev/rmcl/items.py

• class RemarkableCloudManager 64.0% similar

  Unified manager for reMarkable Cloud operations that uses REST API as primary method with rmcl library as fallback, handling authentication, file operations, and folder management.

  From: /tf/active/vicechatdev/e-ink-llm/remarkable_cloud.py

• class RemarkableNode_v1 63.3% similar

  A dataclass representing a node (folder or document) in the reMarkable cloud storage system, storing metadata, hashes, and local file paths.

  From: /tf/active/vicechatdev/e-ink-llm/cloudtest/local_replica_v2.py