160 lines
5.1 KiB
Python
160 lines
5.1 KiB
Python
from typing import Optional, List
|
|
from sqlmodel import Field, Relationship, SQLModel
|
|
|
|
from arcology.parse import parse_sexp
|
|
|
|
class Link(SQLModel, table=True):
|
|
__tablename__ = 'links'
|
|
pos: str
|
|
source_ref: Optional[str] = Field(primary_key=True, foreign_key="nodes.id",
|
|
sa_column_kwargs=dict(name='source'))
|
|
dest_ref: Optional[str] = Field(primary_key=True, foreign_key="nodes.id",
|
|
sa_column_kwargs=dict(name='dest'))
|
|
type: str
|
|
properties: str
|
|
|
|
source: Optional["Node"] = Relationship(back_populates="outlinks",
|
|
sa_relationship_kwargs=dict(
|
|
foreign_keys="[Link.source_ref]",
|
|
lazy="joined",
|
|
))
|
|
dest: Optional["Node"] = Relationship(back_populates="backlinks",
|
|
sa_relationship_kwargs=dict(
|
|
foreign_keys="[Link.dest_ref]",
|
|
lazy="joined",
|
|
))
|
|
|
|
class Node(SQLModel, table=True):
|
|
__tablename__ = 'nodes'
|
|
id: Optional[int] = Field(default=None, primary_key=True)
|
|
level: int
|
|
pos: int
|
|
title: str
|
|
|
|
todo: str
|
|
priority: str
|
|
|
|
scheduled: str
|
|
deadline: str
|
|
|
|
properties: str
|
|
olp: str
|
|
|
|
file_ref: Optional[str] = Field(foreign_key="files.file",
|
|
sa_column_kwargs={'name': 'file'})
|
|
file: List["File"] = Relationship(back_populates="nodes")
|
|
aliases: List["Alias"] = Relationship(back_populates="node")
|
|
tags: List["Tag"] = Relationship(back_populates="node")
|
|
references: List["Reference"] = Relationship(back_populates="node")
|
|
outlinks: List["Link"] = Relationship(
|
|
back_populates="dest",
|
|
sa_relationship_kwargs=dict(
|
|
viewonly=True,
|
|
foreign_keys="[Link.dest_ref]",
|
|
primaryjoin="Node.id==Link.dest_ref",
|
|
))
|
|
|
|
outlink_nodes: List["Node"] = Relationship(
|
|
back_populates="backlink_nodes",
|
|
link_model=Link,
|
|
sa_relationship_kwargs=dict(
|
|
viewonly=True,
|
|
primaryjoin="Node.id==Link.dest_ref",
|
|
secondaryjoin="Node.id==Link.source_ref",
|
|
)
|
|
)
|
|
|
|
backlinks: List["Link"] = Relationship(
|
|
back_populates="source",
|
|
sa_relationship_kwargs=dict(
|
|
viewonly=True,
|
|
foreign_keys="[Link.source_ref]",
|
|
primaryjoin="Node.id==Link.source_ref",
|
|
))
|
|
|
|
# https://github.com/tiangolo/sqlmodel/issues/89
|
|
backlink_nodes: List["Node"] = Relationship(
|
|
back_populates="outlink_nodes",
|
|
link_model=Link,
|
|
sa_relationship_kwargs=dict(
|
|
viewonly=True,
|
|
primaryjoin="Node.id==Link.source_ref",
|
|
secondaryjoin="Node.id==Link.dest_ref",
|
|
)
|
|
)
|
|
|
|
class File(SQLModel, table=True):
|
|
__tablename__ = 'files'
|
|
file: Optional[str] = Field(default=None, primary_key=True)
|
|
hash: str
|
|
atime: str
|
|
mtime: str
|
|
|
|
nodes: List["Node"] = Relationship(back_populates="file")
|
|
keywords: List["Keyword"] = Relationship(back_populates="file")
|
|
|
|
class Alias(SQLModel, table=True):
|
|
__tablename__ = 'aliases'
|
|
node_id: str = Field(primary_key=True, foreign_key='nodes.id')
|
|
alias: str
|
|
|
|
node: Optional["Node"] = Relationship(back_populates="aliases")
|
|
|
|
def __str__(self):
|
|
return parse_sexp(self.alias)
|
|
|
|
class Tag(SQLModel, table=True):
|
|
__tablename__ = 'tags'
|
|
node_id: str = Field(primary_key=True, foreign_key='nodes.id')
|
|
tag: str
|
|
|
|
node: Optional["Node"] = Relationship(back_populates="tags")
|
|
|
|
def __str__(self):
|
|
return parse_sexp(self.tag)
|
|
|
|
class Reference(SQLModel, table=True):
|
|
__tablename__ = 'refs'
|
|
node_id: str = Field(primary_key=True, foreign_key='nodes.id')
|
|
ref: str
|
|
|
|
node: Optional["Node"] = Relationship(back_populates="references")
|
|
|
|
def __str__(self):
|
|
return parse_sexp(self.ref)
|
|
|
|
class Keyword(SQLModel, table=True):
|
|
__tablename__ = 'keywords'
|
|
keyword: str
|
|
value: str = Field(str, primary_key=True)
|
|
|
|
file_ref: Optional[str] = Field(foreign_key="files.file",
|
|
sa_column_kwargs={'name': 'file'},
|
|
primary_key=True)
|
|
file: List["File"] = Relationship(back_populates="keywords")
|
|
|
|
def __init__(self):
|
|
super(self)
|
|
self.__table__.schema = "arroyo"
|
|
|
|
def __str__(self):
|
|
return {keyword: value}
|
|
|
|
from sqlmodel import create_engine
|
|
from sqlalchemy import event
|
|
|
|
org_roam_sqlite_file_name = "/home/rrix/.emacs.d/org-roam.db"
|
|
arroyo_sqlite_file_name = "/home/rrix/.emacs.d/arroyo.db"
|
|
|
|
def make_engine():
|
|
engine = create_engine('sqlite:///{path}'.format(path=org_roam_sqlite_file_name), echo=True)
|
|
|
|
@event.listens_for(engine, "connect")
|
|
def do_connect(dbapi_connection, _connection_record):
|
|
dbapi_connection.execute("attach database '{ar}' as arroyo;".format(ar=arroyo_sqlite_file_name))
|
|
|
|
return engine
|
|
|
|
|
|
engine = make_engine()
|