Вы можете использовать встроенный тип списка, если node
является списком:
node[0] # get by id (id == 0)
node # .all()
Если вы хотите создать специальные классы для реализации желаемого интерфейса, то, написав примерно столько же кода, сколько и для пользовательских классов, вы можете получить полную мощность реляционных баз данных, используя SQLAlchemy :
# get node id
node.id # -> 1
# get all nodes
Node.query.all() # -> [<Node 1>, <Node 2>, <Node 3>, <Node 4>]
# get all nodes associated with the `way`
way.nodes # -> [<Node 2>, <Node 3>, <Node 4>]
# get `way` with id == 1
Way.query.get(1) # -> <Way 1>
# get all `way`s the node belongs to
node.ways.all() # -> []
# get all nodes with longitude == Y
Node.query.filter_by(lon=Y).all() # -> [<Node 1>, <Node 2>]
Вот код, который его реализует:
from sqlalchemy import create_engine, Column, Integer, Float, Table, ForeignKey
from sqlalchemy.orm import scoped_session, sessionmaker, relationship, backref
from sqlalchemy.ext.declarative import declarative_base, declared_attr
# declare models
class Base(declarative_base()):
__abstract__ = True
# provide id for each object
id = Column(Integer, primary_key=True)
@classmethod
def create(cls, *args, **kwargs):
obj = cls(*args, **kwargs)
Session.add(obj) # add to db session
return obj
@declared_attr
def __tablename__(cls):
"""Use lowercased classname for table name."""
return cls.__name__.lower()
def __repr__(self):
return "<{name} {id}>".format(name=self.__class__.__name__, id=self.id)
class Node(Base):
lat = Column(Float)
lon = Column(Float)
def __init__(self, lat, lon):
self.lat = lat
self.lon = lon
# define many-to-many relationship
nodes = Table(
'nodes', Base.metadata,
Column('node_id', Integer, ForeignKey('node.id')),
Column('way_id', Integer, ForeignKey('way.id'))
)
class Way(Base):
nodes = relationship(Node, secondary=nodes,
backref=backref('ways', lazy='dynamic'))
def __init__(self, nodes):
self.nodes = nodes
Тест
# create in memory db
engine = create_engine('sqlite://')
# one db session per thread
Session = scoped_session(sessionmaker(bind=engine))
Base.query = Session.query_property() # allow queries via Model.query
#
Base.metadata.create_all(bind=engine)
X, Y, Z, W = range(4)
node = Node.create(lat=X, lon=Y)
way_nodes = [Node.create(*args) for args in [(X, Y), (Z, W)]]
way = Way.create(way_nodes)
way.nodes.append(Node(10, 11))
Session.commit()
assert node.id == 1
assert node.ways.all() == []
assert Node.query.filter_by(lon=Y).count() == 2
assert way.id == 1
assert any(x.lat == 10 and x.lon == 11 for x in way.nodes)
assert Node.query.filter_by(lat=10, lon=11).one()
assert Way.query.get(1) == way
# remove context (thread) -local session
Session.remove()