Database Models¶

Our RDBMS layer is very interesting: obviously it incorporates all of the traditional elements - all SQLAlchemy, but in order to be resiliant, we’ve a lot of JSON fields. These all have Pydantic models associated with them, so everything is deeply Pythonic.

class lbn.aidoc.models.Comment¶

Bases: SQLModel

__init__(**data)¶

as_str()¶

category: str¶

content: str¶

id: int | None¶

model_config = {'from_attributes': True, 'read_from_attributes': True, 'read_with_orm_mode': True, 'registry': PydanticUndefined, 'table': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property parent¶: Automatically extracts the database session from itself to find the parent model.

property parent_class¶

parent_id: int¶

parent_type: str¶

timestamp: datetime¶

class lbn.aidoc.models.Company¶

Bases: BaseEntity

__init__(**data)¶

analytics: Any¶

as_str()¶

Abstract string serialization method.

Must be explicitly overridden by every child model to pass system tests.

code: ANZSICCode | None¶

property employees: List[Person]¶: The genuine internal team: Bosses, Peers, Talent, etc.

geo: str | None¶

id¶

property is_agency: bool¶: Dynamically classifies recruitment status by reading taxonomy depth.

jobs: Mapped[List[Job]]¶

linkedin: str | None¶

location: str | None¶

model_config = {'from_attributes': True, 'read_from_attributes': True, 'read_with_orm_mode': True, 'registry': PydanticUndefined, 'table': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

property recruiters: List[Person]¶: The external agency partners linked to this company.

state: EntityStatus¶

stints: Mapped[List[EmploymentHistory]]¶

tier: CompanyTier | None¶

website: str | None¶

class lbn.aidoc.models.CompanyAlias¶

Bases: SQLModel

__init__(**data)¶

company_id: int¶

model_config = {'from_attributes': True, 'read_from_attributes': True, 'read_with_orm_mode': True, 'registry': PydanticUndefined, 'table': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

class lbn.aidoc.models.EmploymentHistory¶

Bases: SQLModel

The Ledger: Tracks who works (or worked) where.

__init__(**data)¶

archetype: Archetype¶

company: Mapped[Company]¶

company_id: int¶

end_date: datetime | None¶

id: int | None¶

model_config = {'from_attributes': True, 'read_from_attributes': True, 'read_with_orm_mode': True, 'registry': PydanticUndefined, 'table': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

person: Mapped[Person]¶

person_id: int¶

start_date: datetime | None¶

classmethod validate_styled_enums(v, info)¶: The ‘Inhale Handshake’: Forces Pydantic to use the StyledEnum constructor. This re-attaches the .style metadata lost during AI extraction.

class lbn.aidoc.models.History¶

Bases: SQLModel

for state machine transition histories

__init__(**data)¶

id: int | None¶

model_config = {'from_attributes': True, 'read_from_attributes': True, 'read_with_orm_mode': True, 'registry': PydanticUndefined, 'table': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

new_status: str | None¶

note: str¶

old_status: str | None¶

parent_id: int¶

parent_type: str¶

timestamp: datetime¶

class lbn.aidoc.models.Interview¶

Bases: BaseEntity

__init__(**data)¶

as_ical() → str¶: Surgical Export: Generates a valid iCalendar byte string using the icalendar library.

as_str()¶

Abstract string serialization method.

Must be explicitly overridden by every child model to pass system tests.

attendee_ids: List[int]¶

property attendees: List[Person]¶: Surgical Lookahead: Direct session query to resolve the JSON list of IDs. Returns a list of Person objects matching the internal array registry.

history: Mapped[List[History]]¶

id¶

job: Mapped[Job | None]¶

job_id: int | None¶

location: str¶

model_config = {'from_attributes': True, 'read_from_attributes': True, 'read_with_orm_mode': True, 'registry': PydanticUndefined, 'table': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

predecessor: Mapped[Interview | None]¶

predecessor_id: int | None¶

predecessors() → List[Interview]¶: Walks backwards up the chain of predecessor links. Returns a list of all prior interview objects, sorted chronologically.

scheduled_at: datetime¶

state: MeetingStatus¶

title: str¶

classmethod validate_styled_enums(v, info)¶: The ‘Inhale Handshake’: Forces Pydantic to use the StyledEnum constructor. This re-attaches the .style metadata lost during AI extraction.

class lbn.aidoc.models.Person¶

Bases: BaseEntity

The Master Address Book: Purely digital presence. Independence: No relationships or back-links defined here.

__init__(**data)¶

analytics: Any¶

apikey: str | None¶

as_str()¶

Abstract string serialization method.

Must be explicitly overridden by every child model to pass system tests.

check_password(raw_password: str) → bool¶: Verify an incoming password attempt against the database hash.

colleagues(current_only: bool = True) → List[Person]¶

Returns a list of all distinct people who work (or worked) at the same companies.

Executes a native SQL self-join on the ledger, completely bypassing in-memory loops.

email: str | None¶

property employer: Company | None¶: Returns the current Company from the latest stint.

first_name()¶

generate_apikey() → str¶: Generate a secure, random API prefix string for headless CLI requests.

geo: str | None¶

id¶

jobtitle: str | None¶

last_name()¶

property latest_stint: EmploymentHistory | None¶: Returns the newest stint based on primary key ID fallback.

linkedin: str | None¶

location: str | None¶

model_config = {'from_attributes': True, 'read_from_attributes': True, 'read_with_orm_mode': True, 'registry': PydanticUndefined, 'table': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

password_hash: str | None¶

phone: str | None¶

set_password(raw_password: str) → None¶: Hash a password securely using Python 3.13 standard libraries.

state: EntityStatus¶

stints: Mapped[List[EmploymentHistory]]¶

username: str | None¶

website: str | None¶

class lbn.aidoc.models.PersonAlias¶

Bases: SQLModel

__init__(**data)¶

model_config = {'from_attributes': True, 'read_from_attributes': True, 'read_with_orm_mode': True, 'registry': PydanticUndefined, 'table': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

person_id: int¶

class lbn.aidoc.models.Prompt¶

Bases: SQLModel

an LLM instruction, with well defined inputs and outputs

__init__(**data)¶

id: int | None¶

input_model() → Type[BaseModel]¶: The ‘Inhale Contractor’: Dynamic validation block for input parameters. Recursively compiles a genuine Pydantic model contract straight from the stored input_schema JSON dictionary.

input_schema: Dict[str, Any]¶

inputs() → Dict[str, List[str]]¶: Decomposes the input JSONSchema.

model_config = {'from_attributes': True, 'read_from_attributes': True, 'read_with_orm_mode': True, 'registry': PydanticUndefined, 'table': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

output_model() → Type[BaseModel]¶: The ‘Inhale Contractor’: High-fidelity type mapping. Recursively compiles a genuine, deeply nested Pydantic model contract straight from the stored output_schema JSON dictionary.

output_schema: Dict[str, Any]¶

outputs() → Dict[str, List[str]]¶: Decomposes the output JSONSchema.

provider: str | None¶

rst_files: Mapped[List[RSTFile]]¶

system_message: str¶

temperature: float¶

ttl: int¶

validate_schema_integrity() → Prompt¶: Ensure every required field has a matching property definition.

weight: Weight¶

class lbn.aidoc.models.Prospect¶

Bases: BaseEntity, ABC

Something that we are ingesting and doing a sales/procurment process around.

closes: datetime | None¶

company_id: int | None¶

contact_id: int | None¶

geo: str | None¶

listed: datetime | None¶

location: str | None¶

model_config = {'from_attributes': True, 'ignored_types': (<class 'sqlalchemy.ext.hybrid.hybrid_property'>, <class 'lbn.aidoc.models.linkmanager.LinkManager'>), 'registry': PydanticUndefined}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

source_id: int | None¶

title: str¶

url: str | None¶

class lbn.aidoc.models.RSTFile¶

Bases: SQLModel

__init__(**data)¶

epigraph: str | None¶

epilog: str | None¶

id: int | None¶

is_active: bool¶

last_prompt_id: int | None¶

model_config = {'from_attributes': True, 'read_from_attributes': True, 'read_with_orm_mode': True, 'registry': PydanticUndefined, 'table': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

prolog: str | None¶

prompt: Mapped[Prompt | None]¶

raw_content: str | None¶

seealso: str | None¶

target_path: str¶

class lbn.aidoc.models.Source¶

Bases: SQLModel

__init__(**data)¶

domain: str¶

driver: str¶

extras: Dict[str, Any]¶

classmethod get_by_url(session: Session, url: str) → Source | None¶

The Discovery Drill: Resolves Source by database scheme indexing or domain regular expression cascades.

PRINCIPAL: Returns a Source object, None for custom tool fallbacks,: or raises SyntaxError for unmapped web traffic.

id: int | None¶

jobs: Mapped[List[Job]]¶

model_config = {'from_attributes': True, 'read_from_attributes': True, 'read_with_orm_mode': True, 'registry': PydanticUndefined, 'table': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str¶

password: str | None¶

scheme: str | None¶

searches: Dict[str, Any]¶

url: str | None¶

username: str | None¶

class lbn.aidoc.models.StyledType¶

Bases: TypeDecorator

The Bridge: Handles DB round-trips for any StyledEnum.

__init__(enum_class, *args, **kwargs)¶

Construct a TypeDecorator.

Arguments sent here are passed to the constructor of the class assigned to the impl class level attribute, assuming the impl is a callable, and the resulting object is assigned to the self.impl instance attribute (thus overriding the class attribute of the same name).

If the class level impl is not a callable (the unusual case), it will be assigned to the same instance attribute ‘as-is’, ignoring those arguments passed to the constructor.

Subclasses can override this to customize the generation of self.impl entirely.

cache_ok = True¶

Indicate if statements using this ExternalType are “safe to cache”.

The default value None will emit a warning and then not allow caching of a statement which includes this type. Set to False to disable statements using this type from being cached at all without a warning. When set to True, the object’s class and selected elements from its state will be used as part of the cache key. For example, using a TypeDecorator:

class MyType(TypeDecorator):
    impl = String

    cache_ok = True

    def __init__(self, choices):
        self.choices = tuple(choices)
        self.internal_only = True

The cache key for the above type would be equivalent to:

>>> MyType(["a", "b", "c"])._static_cache_key
(<class '__main__.MyType'>, ('choices', ('a', 'b', 'c')))

The caching scheme will extract attributes from the type that correspond to the names of parameters in the __init__() method. Above, the “choices” attribute becomes part of the cache key but “internal_only” does not, because there is no parameter named “internal_only”.

The requirements for cacheable elements is that they are hashable and also that they indicate the same SQL rendered for expressions using this type every time for a given cache value.

To accommodate for datatypes that refer to unhashable structures such as dictionaries, sets and lists, these objects can be made “cacheable” by assigning hashable structures to the attributes whose names correspond with the names of the arguments. For example, a datatype which accepts a dictionary of lookup values may publish this as a sorted series of tuples. Given a previously un-cacheable type as:

class LookupType(UserDefinedType):
    """a custom type that accepts a dictionary as a parameter.

    this is the non-cacheable version, as "self.lookup" is not
    hashable.

    """

    def __init__(self, lookup):
        self.lookup = lookup

    def get_col_spec(self, **kw):
        return "VARCHAR(255)"

    def bind_processor(self, dialect): ...  # works with "self.lookup" ...

Where “lookup” is a dictionary. The type will not be able to generate a cache key:

>>> type_ = LookupType({"a": 10, "b": 20})
>>> type_._static_cache_key
<stdin>:1: SAWarning: UserDefinedType LookupType({'a': 10, 'b': 20}) will not
produce a cache key because the ``cache_ok`` flag is not set to True.
Set this flag to True if this type object's state is safe to use
in a cache key, or False to disable this warning.
symbol('no_cache')

If we did set up such a cache key, it wouldn’t be usable. We would get a tuple structure that contains a dictionary inside of it, which cannot itself be used as a key in a “cache dictionary” such as SQLAlchemy’s statement cache, since Python dictionaries aren’t hashable:

>>> # set cache_ok = True
>>> type_.cache_ok = True

>>> # this is the cache key it would generate
>>> key = type_._static_cache_key
>>> key
(<class '__main__.LookupType'>, ('lookup', {'a': 10, 'b': 20}))

>>> # however this key is not hashable, will fail when used with
>>> # SQLAlchemy statement cache
>>> some_cache = {key: "some sql value"}
Traceback (most recent call last): File "<stdin>", line 1,
in <module> TypeError: unhashable type: 'dict'

The type may be made cacheable by assigning a sorted tuple of tuples to the “.lookup” attribute:

class LookupType(UserDefinedType):
    """a custom type that accepts a dictionary as a parameter.

    The dictionary is stored both as itself in a private variable,
    and published in a public variable as a sorted tuple of tuples,
    which is hashable and will also return the same value for any
    two equivalent dictionaries.  Note it assumes the keys and
    values of the dictionary are themselves hashable.

    """

    cache_ok = True

    def __init__(self, lookup):
        self._lookup = lookup

        # assume keys/values of "lookup" are hashable; otherwise
        # they would also need to be converted in some way here
        self.lookup = tuple((key, lookup[key]) for key in sorted(lookup))

    def get_col_spec(self, **kw):
        return "VARCHAR(255)"

    def bind_processor(self, dialect): ...  # works with "self._lookup" ...

Where above, the cache key for LookupType({"a": 10, "b": 20}) will be:

>>> LookupType({"a": 10, "b": 20})._static_cache_key
(<class '__main__.LookupType'>, ('lookup', (('a', 10), ('b', 20))))

Added in version 1.4.14: - added the cache_ok flag to allow some configurability of caching for TypeDecorator classes.

Added in version 1.4.28: - added the ExternalType mixin which generalizes the cache_ok flag to both the TypeDecorator and UserDefinedType classes.