julee.repositories.memory

Memory repository implementations for julee domain.

This module exports in-memory implementations of all repository protocols for the Capture, Extract, Assemble, Publish workflow. These implementations use Python dictionaries for storage and are ideal for testing scenarios where external dependencies should be avoided.

All implementations maintain the same async interfaces as their production counterparts while providing lightweight, dependency-free alternatives.

Submodules

• julee.repositories.memory.assembly
• julee.repositories.memory.assembly_specification
• julee.repositories.memory.base
• julee.repositories.memory.document
• julee.repositories.memory.document_policy_validation
• julee.repositories.memory.knowledge_service_config
• julee.repositories.memory.knowledge_service_query
• julee.repositories.memory.policy

Classes

MemoryAssemblyRepository	Memory implementation of AssemblyRepository using Python dictionaries.
MemoryAssemblySpecificationRepository	Memory implementation of AssemblySpecificationRepository using Python
MemoryDocumentPolicyValidationRepository	Memory implementation of DocumentPolicyValidationRepository using Python
MemoryDocumentRepository	Memory implementation of DocumentRepository using Python dictionaries.
MemoryKnowledgeServiceConfigRepository	Memory implementation of KnowledgeServiceConfigRepository using Python
MemoryKnowledgeServiceQueryRepository	Memory implementation of KnowledgeServiceQueryRepository using Python
MemoryPolicyRepository	Memory implementation of PolicyRepository using Python dictionaries.

Package Contents

class julee.repositories.memory.MemoryAssemblyRepository

Bases: julee.domain.repositories.assembly.AssemblyRepository, julee.repositories.memory.base.MemoryRepositoryMixin[julee.domain.models.assembly.Assembly]

Memory implementation of AssemblyRepository using Python dictionaries.

This implementation stores assembly data in memory using a dictionary keyed by assembly_id. This provides a lightweight, dependency-free option for testing.

async generate_id()

Generate a unique assembly identifier.

Returns:

Unique assembly ID string

Return type:

str

async get(assembly_id)

Retrieve an assembly by ID.

Parameters:

assembly_id (str) – Unique assembly identifier

Returns:

Assembly if found, None otherwise

Return type:

julee.domain.models.assembly.Assembly | None

async get_many(assembly_ids)

Retrieve multiple assemblies by ID.

Parameters:

assembly_ids (list[str]) – List of unique assembly identifiers

Returns:

Dict mapping assembly_id to Assembly (or None if not found)

Return type:

dict[str, julee.domain.models.assembly.Assembly | None]

async save(assembly)

Save assembly metadata (status, updated_at, etc.).

Parameters:

assembly (julee.domain.models.assembly.Assembly) – Assembly entity

entity_name = 'Assembly'

logger

storage_dict: dict[str, julee.domain.models.assembly.Assembly]

class julee.repositories.memory.MemoryAssemblySpecificationRepository

Bases: julee.domain.repositories.assembly_specification.AssemblySpecificationRepository, julee.repositories.memory.base.MemoryRepositoryMixin[julee.domain.models.assembly_specification.AssemblySpecification]

Memory implementation of AssemblySpecificationRepository using Python dictionaries.

This implementation stores assembly specifications in memory:

• Specifications: Dictionary keyed by assembly_specification_id containing AssemblySpecification objects

This provides a lightweight, dependency-free option for testing while maintaining the same interface as other implementations.

async generate_id()

Generate a unique assembly specification identifier.

Returns:

Unique assembly specification ID string

Return type:

str

async get(assembly_specification_id)

Retrieve an assembly specification by ID.

Parameters:

assembly_specification_id (str) – Unique specification identifier

Returns:

AssemblySpecification if found, None otherwise

Return type:

julee.domain.models.assembly_specification.AssemblySpecification | None

async get_many(assembly_specification_ids)

Retrieve multiple assembly specifications by ID.

Parameters:

• assembly_specification_ids (list[str]) – List of unique specification

• identifiers

Returns:

Dict mapping specification_id to AssemblySpecification (or None if not found)

Return type:

dict[str, julee.domain.models.assembly_specification.AssemblySpecification | None]

async list_all()

List all assembly specifications.

Returns:

List of all AssemblySpecification entities in the repository

Return type:

list[julee.domain.models.assembly_specification.AssemblySpecification]

async save(assembly_specification)

Save an assembly specification.

Parameters:

assembly_specification (julee.domain.models.assembly_specification.AssemblySpecification) – Complete AssemblySpecification to save

entity_name = 'AssemblySpecification'

logger

storage_dict: dict[str, julee.domain.models.assembly_specification.AssemblySpecification]

class julee.repositories.memory.MemoryDocumentPolicyValidationRepository

Bases: julee.domain.repositories.document_policy_validation.DocumentPolicyValidationRepository, julee.repositories.memory.base.MemoryRepositoryMixin[julee.domain.models.policy.DocumentPolicyValidation]

Memory implementation of DocumentPolicyValidationRepository using Python dictionaries.

This implementation stores document policy validation data in memory using a dictionary keyed by validation_id. This provides a lightweight, dependency-free option for testing.

async generate_id()

Generate a unique validation identifier.

Returns:

Unique validation ID string

Return type:

str

async get(validation_id)

Retrieve a document policy validation by ID.

Parameters:

validation_id (str) – Unique validation identifier

Returns:

DocumentPolicyValidation if found, None otherwise

Return type:

julee.domain.models.policy.DocumentPolicyValidation | None

async get_many(validation_ids)

Retrieve multiple document policy validations by ID.

Parameters:

validation_ids (list[str]) – List of unique validation identifiers

Returns:

Dict mapping validation_id to DocumentPolicyValidation (or None if not found)

Return type:

dict[str, julee.domain.models.policy.DocumentPolicyValidation | None]

async save(validation)

Save a document policy validation.

Parameters:

validation (julee.domain.models.policy.DocumentPolicyValidation) – Complete DocumentPolicyValidation to save

entity_name = 'DocumentPolicyValidation'

logger

storage_dict: dict[str, julee.domain.models.policy.DocumentPolicyValidation]

class julee.repositories.memory.MemoryDocumentRepository

Bases: julee.domain.repositories.document.DocumentRepository, julee.repositories.memory.base.MemoryRepositoryMixin[julee.domain.models.document.Document]

Memory implementation of DocumentRepository using Python dictionaries.

This implementation stores document metadata and content in memory: - Documents: Dictionary keyed by document_id containing Document objects

This provides a lightweight, dependency-free option for testing while maintaining the same interface as other implementations.

async generate_id()

Generate a unique document identifier.

Returns:

Unique document ID string

Return type:

str

async get(document_id)

Retrieve a document with metadata and content.

Parameters:

document_id (str) – Unique document identifier

Returns:

Document object if found, None otherwise

Return type:

julee.domain.models.document.Document | None

async get_many(document_ids)

Retrieve multiple documents by ID.

Parameters:

document_ids (list[str]) – List of unique document identifiers

Returns:

Dict mapping document_id to Document (or None if not found)

Return type:

dict[str, julee.domain.models.document.Document | None]

async list_all()

List all documents.

Returns:

List of all Document entities in the repository

Return type:

list[julee.domain.models.document.Document]

async save(document)

Save a document with its content and metadata.

If the document has content_bytes, it will be normalized to bytes (encoding str as UTF-8), converted to a ContentStream and the content hash will be calculated automatically.

Parameters:

document (julee.domain.models.document.Document) – Document object to save

Raises:

• ValueError – If document has no content or content_bytes

• TypeError – If content_bytes is not bytes or str

entity_name = 'Document'

logger

storage_dict: dict[str, julee.domain.models.document.Document]

class julee.repositories.memory.MemoryKnowledgeServiceConfigRepository

Bases: julee.domain.repositories.knowledge_service_config.KnowledgeServiceConfigRepository, julee.repositories.memory.base.MemoryRepositoryMixin[julee.domain.models.knowledge_service_config.KnowledgeServiceConfig]

Memory implementation of KnowledgeServiceConfigRepository using Python dictionaries.

This implementation stores knowledge service configurations in memory:

• Knowledge Services: Dictionary keyed by knowledge_service_id containing KnowledgeServiceConfig objects

This provides a lightweight, dependency-free option for testing while maintaining the same interface as other implementations.

async generate_id()

Generate a unique knowledge service identifier.

Returns:

Unique knowledge service ID string

Return type:

str

async get(knowledge_service_id)

Retrieve a knowledge service configuration by ID.

Parameters:

knowledge_service_id (str) – Unique knowledge service identifier

Returns:

KnowledgeServiceConfig object if found, None otherwise

Return type:

julee.domain.models.knowledge_service_config.KnowledgeServiceConfig | None

async get_many(knowledge_service_ids)

Retrieve multiple knowledge service configs by ID.

Parameters:

• knowledge_service_ids (list[str]) – List of unique knowledge service

• identifiers

Returns:

Dict mapping knowledge_service_id to KnowledgeServiceConfig (or None if not found)

Return type:

dict[str, julee.domain.models.knowledge_service_config.KnowledgeServiceConfig | None]

async list_all()

List all knowledge service configurations.

Returns:

List of all KnowledgeServiceConfig entities in the repository

Return type:

list[julee.domain.models.knowledge_service_config.KnowledgeServiceConfig]

async save(knowledge_service)

Save a knowledge service configuration.

Parameters:

knowledge_service (julee.domain.models.knowledge_service_config.KnowledgeServiceConfig) – Complete KnowledgeServiceConfig to save

entity_name = 'KnowledgeServiceConfig'

logger

storage_dict: dict[str, julee.domain.models.knowledge_service_config.KnowledgeServiceConfig]

class julee.repositories.memory.MemoryKnowledgeServiceQueryRepository

Bases: julee.domain.repositories.knowledge_service_query.KnowledgeServiceQueryRepository, julee.repositories.memory.base.MemoryRepositoryMixin[julee.domain.models.assembly_specification.KnowledgeServiceQuery]

Memory implementation of KnowledgeServiceQueryRepository using Python dictionaries.

This implementation stores knowledge service queries in memory:

• Queries: Dictionary keyed by query_id containing KnowledgeServiceQuery objects

This provides a lightweight, dependency-free option for testing while maintaining the same interface as other implementations.

async generate_id()

Generate a unique query identifier.

Returns:

Unique string identifier for a new query

Return type:

str

async get(query_id)

Retrieve a knowledge service query by ID.

Parameters:

query_id (str) – Unique query identifier

Returns:

KnowledgeServiceQuery object if found, None otherwise

Return type:

julee.domain.models.assembly_specification.KnowledgeServiceQuery | None

async get_many(query_ids)

Retrieve multiple knowledge service queries by ID.

Parameters:

query_ids (list[str]) – List of unique query identifiers

Returns:

Dict mapping query_id to KnowledgeServiceQuery (or None if not found)

Return type:

dict[str, julee.domain.models.assembly_specification.KnowledgeServiceQuery | None]

async list_all()

List all knowledge service queries.

Returns:

List of all knowledge service queries, sorted by query_id

Return type:

list[julee.domain.models.assembly_specification.KnowledgeServiceQuery]

async save(query)

Store or update a knowledge service query.

Parameters:

query (julee.domain.models.assembly_specification.KnowledgeServiceQuery) – KnowledgeServiceQuery object to store

entity_name = 'KnowledgeServiceQuery'

logger

storage_dict: dict[str, julee.domain.models.assembly_specification.KnowledgeServiceQuery]

class julee.repositories.memory.MemoryPolicyRepository

Bases: julee.domain.repositories.policy.PolicyRepository, julee.repositories.memory.base.MemoryRepositoryMixin[julee.domain.models.policy.Policy]

Memory implementation of PolicyRepository using Python dictionaries.

This implementation stores policy data in memory using a dictionary keyed by policy_id. This provides a lightweight, dependency-free option for testing.

async generate_id()

Generate a unique policy identifier.

Returns:

Unique policy ID string

Return type:

str

async get(policy_id)

Retrieve a policy by ID.

Parameters:

policy_id (str) – Unique policy identifier

Returns:

Policy if found, None otherwise

Return type:

julee.domain.models.policy.Policy | None

async get_many(policy_ids)

Retrieve multiple policies by ID.

Parameters:

policy_ids (list[str]) – List of unique policy identifiers

Returns:

Dict mapping policy_id to Policy (or None if not found)

Return type:

dict[str, julee.domain.models.policy.Policy | None]

async save(policy)

Save a policy.

Parameters:

policy (julee.domain.models.policy.Policy) – Complete Policy to save

entity_name = 'Policy'

logger

storage_dict: dict[str, julee.domain.models.policy.Policy]