Source code for pygtfs.schedule

from __future__ import (division, absolute_import, print_function,

import sqlalchemy
import sqlalchemy.orm

from .gtfs_entities import gtfs_all, Feed, Base

[docs]class Schedule: """Represents the full database. The schedule is the most important object in pygtfs. It represents the entire dataset. Most of the properties come straight from the gtfs reference. Two of them were renamed: calendar is called `services`, and calendar_dates `service_exceptions`. One addition is the `feeds` table, which is here to support more than one feed in a database. Each of the properties is a list created upon access by sqlalchemy. Then, each element of the list as attributes following the gtfs reference. In addition, if they are related to another table, this can also be accessed by attribute. :param db_conection: Either a sqlalchemy database url or a filename to be used with sqlite. """ def __init__(self, db_connection): self.db_connection = db_connection self.db_filename = None if '://' not in db_connection: self.db_connection = 'sqlite:///%s' % self.db_connection if self.db_connection.startswith('sqlite'): self.db_filename = self.db_connection self.engine = sqlalchemy.create_engine(self.db_connection, echo=False) Session = sqlalchemy.orm.sessionmaker(bind=self.engine) self.session = Session() Base.metadata.create_all(self.engine)
[docs] def drop_feed(self, feed_id): """ Delete a feed from a database by feed id""" # the following does not cascade unfortunatly. # self.session.query(Feed).filter(Feed.feed_id == feed_id).delete() feed = self.session.query(Feed).get(feed_id) self.session.delete(feed) self.session.commit()
def _meta_query_all(entity, docstring=None): def _query_all(instance_self): """ A list generated on access """ return instance_self.session.query(entity).all() if docstring is not None: _query_all.__doc__ = docstring return property(_query_all) def _meta_query_by_id(entity, docstring=None): def _query_by_id(self, id): """ A function that returns a list of entries with matching ids """ return self.session.query(entity).filter( == id).all() if docstring is not None: _query_by_id.__doc__ = docstring return _query_by_id def _meta_query_raw(entity, docstring=None): def _query_raw(instance_self): """ A raw sqlalchemy query object that the user can then manipulate manually """ return instance_self.session.query(entity) if docstring is not None: _query_raw.__doc__ = docstring return property(_query_raw) for entity in (gtfs_all + [Feed]): entity_doc = "A list of :py:class:`pygtfs.gtfs_entities.{0}` objects".format(entity.__name__) entity_raw_doc = ("A :py:class:`sqlalchemy.orm.Query` object to fetch " ":py:class:`pygtfs.gtfs_entities.{0}` objects" .format(entity.__name__)) entity_by_id_doc = "A list of :py:class:`pygtfs.gtfs_entities.{0}` objects with matching id".format(entity.__name__) setattr(Schedule, entity._plural_name_, _meta_query_all(entity, entity_doc)) setattr(Schedule, entity._plural_name_ + "_query", _meta_query_raw(entity, entity_raw_doc)) if hasattr(entity, 'id'): setattr(Schedule, entity._plural_name_ + "_by_id", _meta_query_by_id(entity, entity_by_id_doc))