init

Backend only schemas (not used in any way by the frontend).

I.e. These can be used to work with data that is loaded from the database by a related SqlAlchemy model where it is never necessary to send the full data to the frontend. Information from these schema might be used to populate schemas that are shared with the frontend.

DatabaseSchemaType = TypeVar('DatabaseSchemaType', bound=DatabaseSchemaMixin | SchemaBase) module-attribute

NOT_LOADED = NotLoaded() module-attribute

ValidatedUTCDatetime = Annotated[datetime.datetime, AfterValidator(_assert_utc)] module-attribute

ValidatedUUID = Annotated[UUID, BeforeValidator(_make_valid_uuid)] module-attribute

__all__ = ['NOT_LOADED', 'AgentSchema', 'AssistantRunSchema', 'AssistantRunStatuses', 'AssistantSchema', 'BasicUserInfo', 'ClassChunk', 'CodeFileChunks', 'ConfigID', 'ConfigType', 'ConversationMetadataSchema', 'DatabaseSchemaMixin', 'DatabaseSchemaType', 'DecoratorChunk', 'FunctionCallSchema', 'FunctionChunk', 'FunctionResponseSchema', 'GithubMetadata', 'ImportChunk', 'LLMModels', 'MessageRoles', 'MessageSchema', 'NotLoaded', 'OptionsSchema', 'ParsedCodeFile', 'ParsedGithubFile', 'ParsedRepositoryInfo', 'ParsedTextFile', 'PurchaseSchema', 'SchemaBase', 'UpdatableProtocol', 'Update', 'UsageSchema', 'UserRoles', 'ValidatedUTCDatetime', 'ValidatedUUID', 'ensure_uuid'] module-attribute

AgentSchema

Bases: ConfigSchemaBase

agent_type = Field(default='basic', description='For differentiating structurally different Agents (e.g. different prompt structures)') class-attribute instance-attribute

options = OptionsSchema() class-attribute instance-attribute

sub_assistants = Field(default_factory=list) class-attribute instance-attribute

tools = Field(default_factory=list) class-attribute instance-attribute

AssistantRunSchema

Bases: DatabaseSchemaMixin, SchemaBase

Schema for an assistant run.

An assistant run is a user triggered run of an assistant (that could include sub-assistant runs).

assistant_id instance-attribute

contexts_used = Field(default_factory=list) class-attribute instance-attribute

conversation_uid instance-attribute

conversation_version instance-attribute

created_at = Field(default_factory=lambda: datetime.datetime.now(datetime.UTC)) class-attribute instance-attribute

current_step = 0 class-attribute instance-attribute

error_message = None class-attribute instance-attribute

finished_at = None class-attribute instance-attribute

new_instance_on_change = False class-attribute

new_messages = Field(default_factory=list) class-attribute instance-attribute

partial_info = None class-attribute instance-attribute

previous_thread_uid instance-attribute

run_index instance-attribute

status = AssistantRunStatuses.NOT_STARTED class-attribute instance-attribute

thread_uid = Field(default_factory=uuid.uuid4) class-attribute instance-attribute

user_id instance-attribute

set_ids_from_model(model)

AssistantRunStatuses

Bases: StrEnum

Enum for request statuses.

COMPLETE = 'complete' class-attribute instance-attribute

ERRORED = 'errored' class-attribute instance-attribute

NOT_STARTED = 'not_started' class-attribute instance-attribute

PARTIAL = 'in_progress' class-attribute instance-attribute

AssistantSchema

Bases: ConfigSchemaBase

agent = AgentSchema() class-attribute instance-attribute

as_tool_description = None class-attribute instance-attribute

langgraph_settings = LanggraphSettingsSchema() class-attribute instance-attribute

BasicUserInfo

Bases: SchemaBase

Minimal version of User for use with Auth state etc.

clerk_id instance-attribute

credits instance-attribute

id instance-attribute

is_admin property

roles instance-attribute

from_user(user) classmethod

ClassChunk

Bases: DatabaseSchemaMixin, SchemaBase

decorators instance-attribute

docstring instance-attribute

full_content instance-attribute

init instance-attribute

iterate_nested_schemas = True class-attribute

line_numbers instance-attribute

methods instance-attribute

name instance-attribute

new_instance_on_change = False class-attribute

signature instance-attribute

superclasses instance-attribute

as_artifact()

as_content()

CodeFileChunks

Bases: DatabaseSchemaMixin, SchemaBase

classes instance-attribute

full_content instance-attribute

functions instance-attribute

imports instance-attribute

iterate_nested_schemas = True class-attribute

language instance-attribute

new_instance_on_change = False class-attribute

num_lines instance-attribute

ConfigID

Bases: UniqueIDBase[ConfigSchemaBase]

name instance-attribute

schema_class instance-attribute

type_ instance-attribute

user_id = None class-attribute instance-attribute

__hash__()

__repr__()

copy_without_version()

equal_excluding_version_and_deletion(other)

from_meta(meta, schema_class, version, deleted) classmethod

from_schema(schema, exact_version=False) classmethod

ConfigType

Bases: StrEnum

GLOBAL = 'global' class-attribute instance-attribute

IMMUTABLE = 'immutable' class-attribute instance-attribute

NOT_SET = 'not_set' class-attribute instance-attribute

USER = 'user' class-attribute instance-attribute

ConversationMetadataSchema

Bases: DatabaseSchemaMixin, SchemaBase

assistant_runs = Field(default_factory=list) class-attribute instance-attribute

conversation_uid instance-attribute

created_at = Field(default_factory=lambda: datetime.datetime.now(datetime.UTC)) class-attribute instance-attribute

default_assistant instance-attribute

iterate_nested_schemas = False class-attribute

last_updated_at = Field(default_factory=lambda: datetime.datetime.now(datetime.UTC)) class-attribute instance-attribute

message_history = Field(default_factory=list) class-attribute instance-attribute

new_instance_on_change = True class-attribute

next_run_index = 0 class-attribute instance-attribute

pending_assistant_runs = Field(default_factory=list) class-attribute instance-attribute

sub_assistant_run_info = Field(default_factory=SubAssistantRunInfo) class-attribute instance-attribute

title instance-attribute

user_id instance-attribute

increment_version()

new(user_pk_id, title, assistant_schema) classmethod

set_initial_version()

DatabaseSchemaMixin

Bases: BaseModel

Adds methods and values for storing in database (including id, version, deleted).

deleted = Field(default=False, exclude=False, description='Whether the schema has marked as deleted') class-attribute instance-attribute

id = Field(default=None, exclude=False, description='Database ID') class-attribute instance-attribute

iterate_nested_schemas = False class-attribute

Should nested schemas (subclasses of DatabaseSchemaMixin) be iterated over for various operations. Defaults to False for efficiency. Also should be left False if the subclass handles behaviour explicitly.

new_instance_on_change = True class-attribute

Should a whole new instance be created in the database for any change (version increment) vs changing in place.

private_incremented = False class-attribute instance-attribute

version = Field(default=None, exclude=False, description='Version of the schema') class-attribute instance-attribute

determine_version_increment(other)

Increment version for any updated schemas or sub-schemas.

I.e., 'self' is the updated schema to store in the database, and 'other' is what currently exists there.

equal_to(other, keys_to_recursively_remove=None, exclude_id=True, include_version=False)

For more complex comparisons where schema1 == schema2 is not sufficient.

E.g. Trying to compare all values without metadatas or ids.

increment_version()

Increment the version number.

Mark that model has been updated since last load. If self.id is set to None (after this) a new entry will be saved in the database.

Only makes sense to do this if the schema has an ID already.

mark_deleted(apply_nested=False)

reset_ids()

Set all ids to None.

Remove all persistence ids.

reset_incremented_for_testing()

Reset the incremented flag.

Only for testing purposes (where several updates to a schema are made within a single UOW).

For normal use, there should only ever be a single update to a schema within a UOW. Allowing multiple updates in a single UOW would allow multiple simultaneous UOWs to update the same record without clashing on version (e.g. if one UOW updates once and one twice before committing, there will be no apparent clash in version).

reset_version(nested)

Set all versions to None.

set_ids_from_model(model)

Set the ID from a model instance.

Used when creating a new schema from a model instance. Only used in repository/uow methods.

Note: Modified in place because it's safer for any schemas that ARE now associated with a saved model to have their ids set to avoid future accidental duplication.

set_initial_version()

DecoratorChunk

Bases: DatabaseSchemaMixin, SchemaBase

args instance-attribute

name instance-attribute

new_instance_on_change = False class-attribute

FunctionCallSchema

Bases: SchemaBase, DatabaseSchemaMixin

args_dict property

arguments instance-attribute

call_id instance-attribute

call_repr property

index instance-attribute

name instance-attribute

__str__()

FunctionChunk

Bases: DatabaseSchemaMixin, SchemaBase

code_body instance-attribute

decorators instance-attribute

docstring instance-attribute

full_content instance-attribute

iterate_nested_schemas = True class-attribute

line_numbers instance-attribute

name instance-attribute

new_instance_on_change = False class-attribute

return_type instance-attribute

signature instance-attribute

as_artifact()

as_content()

FunctionResponseSchema

Bases: SchemaBase, DatabaseSchemaMixin

artifact = None class-attribute instance-attribute

artifact_pk_id = None class-attribute instance-attribute

artifact_tablename = None class-attribute instance-attribute

call_id instance-attribute

name instance-attribute

__str__()

GithubMetadata

Bases: DatabaseSchemaMixin, SchemaBase

branch instance-attribute

node_id instance-attribute

owner instance-attribute

path instance-attribute

repo instance-attribute

size_bytes instance-attribute

as_content()

from_repo_and_blob(repo_id, blob) classmethod

update_from_repo_id(repo_id)

ImportChunk

Bases: DatabaseSchemaMixin, SchemaBase

aliased_name instance-attribute

full_content instance-attribute

import_path instance-attribute

is_relative instance-attribute

line_numbers instance-attribute

name instance-attribute

new_instance_on_change = False class-attribute

relative_levels instance-attribute

as_artifact()

as_content()

LLMModels

Bases: SimpleNamespace

anthropic = AnthropicModels class-attribute instance-attribute

fake = FakeModels class-attribute instance-attribute

openai = OpenaiModels class-attribute instance-attribute

MessageRoles

Bases: Enum

The different roles a message can have (that are understood by openai).

ASSISTANT = 'assistant' class-attribute instance-attribute

SYSTEM = 'system' class-attribute instance-attribute

TOOL_CALL = 'tool_call' class-attribute instance-attribute

TOOL_RESPONSE = 'tool_response' class-attribute instance-attribute

USER = 'user' class-attribute instance-attribute

MessageSchema

Bases: SchemaBase, DatabaseSchemaMixin

content instance-attribute

created_at = Field(default_factory=lambda: datetime.datetime.now(datetime.UTC)) class-attribute instance-attribute

lc_id = None class-attribute instance-attribute

role instance-attribute

tool_calls = Field(default_factory=list) class-attribute instance-attribute

tool_response = None class-attribute instance-attribute

uid = Field(default_factory=uuid.uuid4) class-attribute instance-attribute

__str__()

from_lc(lc_message) classmethod

set_ids_from_model(model)

NotLoaded

Bases: BaseModel

Placeholder for a relationship that has not been loaded.

Use this as a type hint for any attributes that correspond to relationships that may not be loaded (i.e. lazy='raise' in the relationship) This is to distinguish between None compared to a relationship that does exist but just wasn't loaded for this instance.

The UNLOADED instance below should be set as the value for any unloaded relationships

OptionsSchema

Bases: ConfigSchemaBase

Schema for storing all option values in control panel.

condense_settings = CondenseSettingsSchema() class-attribute instance-attribute

context_settings = ContextSettingsSchema() class-attribute instance-attribute

llm_settings = LLMSettingsSchema() class-attribute instance-attribute

personality_settings = PersonalitySettingsSchema() class-attribute instance-attribute

ParsedCodeFile

Bases: ParsedGithubFile

clerk_owner instance-attribute

code_file_chunks instance-attribute

full_content instance-attribute

github_metadata instance-attribute

iterate_nested_schemas = True class-attribute

new_instance_on_change = False class-attribute

from_chunks_and_metadata(chunks, metadata, clerk_owner) classmethod

ParsedGithubFile

Bases: DatabaseSchemaMixin, SchemaBase, ABC

clerk_owner = Field(description="The owner of the repository (either clerk_id or 'public')") class-attribute instance-attribute

full_content = Field(description='The full content of the file.') class-attribute instance-attribute

github_metadata instance-attribute

ParsedRepositoryInfo

Bases: BaseModel

Summarize info about parsed files stored in db.

num_code_files instance-attribute

num_text_files instance-attribute

ParsedTextFile

Bases: ParsedGithubFile

clerk_owner instance-attribute

full_content instance-attribute

general_text_chunks instance-attribute

github_metadata instance-attribute

iterate_nested_schemas = True class-attribute

new_instance_on_change = False class-attribute

from_chunks_and_metadata(chunks, metadata, clerk_owner, full_content) classmethod

PurchaseSchema

Bases: DatabaseSchemaMixin, SchemaBase

checkout_uid instance-attribute

currency = None class-attribute instance-attribute

dollar_amount = None class-attribute instance-attribute

order_complete = False class-attribute instance-attribute

raw_amount = None class-attribute instance-attribute

stripe_item_id instance-attribute

SchemaBase

Bases: BaseModel, ABC

Base class for all schemas.

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

__eq__(other)

Compare by the model_dump so that excluded values are not compared.

equal_to(other, keys_to_recursively_remove=None)

For more complex comparisons where schema1 == schema2 is not sufficient.

Note 2024-05-16: Not easy to compare when some fields are not loaded. Not sure how to handle that yet.

---

other: The other schema to compare to
keys_to_recursively_remove: List of keys to recursively remove from the comparison before checking equality

set_not_loaded_on_unsaved(values)

Set any values that have a NotSaved annotation and are not present to NotLoaded.

UpdatableProtocol

Bases: Protocol

Protocol for an async callable that can accept updates.

Examples:

async def example_updatable(updates: list[Update]) -> None:
    print("Applying updates", updates)

__call__(updates) async

Update dataclass

Base class for updates to propagate up to Controller/Frontend/Anything else.

Subclass to add specific fields.

__init__()

UsageSchema

Bases: SchemaBase, DatabaseSchemaMixin

Pydantic model equivalent of the UsageModel.

completion_tokens instance-attribute

conversation_id instance-attribute

prompt_tokens instance-attribute

total_cost instance-attribute

total_tokens instance-attribute

__add__(other)

empty(conversation_id) classmethod

UserRoles

Bases: Enum

ADMIN = 'ADMIN' class-attribute instance-attribute

BASIC = 'BASIC' class-attribute instance-attribute

PREMIUM = 'PREMIUM' class-attribute instance-attribute

PUBLIC = 'PUBLIC' class-attribute instance-attribute

ensure_uuid(uuid_or_str)

Ensures that a UUID is returned from either a string or a UUID.

Mostly necessary because sqlite doesn't have a native UUID type so it converts UUIDs to strings on commit.

Note: loading a model still returns UUID as appropriate, it's only when accessing variables after commit/flush that is a problem Args: uuid_or_str: