Commits

March Liu  committed ccd0cac

use socped session

  • Participants
  • Parent commits 38eb8db

Comments (0)

Files changed (1)

File src/socrates/core.py

 # -*- coding:utf-8 -*-
 
 from sqlalchemy import create_engine, MetaData, Sequence, Table, Column, String, select, join, Integer, and_
-from sqlalchemy.orm import sessionmaker, mapper
+from sqlalchemy.orm import sessionmaker, scoped_session, mapper
 import meta
 from socrates.types import Subject, Segment, Predicate, Type
 
 >>> storage.registed_type.keys()
 ['string', 'subject']
     """
-    def __init__(self, uri=None, create=False, schema=None):
+    def __init__(self, uri=None, create=False, schema=None, pool_size=20):
         """
 
         """
-        engine = create_engine(uri) if uri else create_engine("sqlite://:memory:")
+        engine = create_engine(uri, pool_size=pool_size) if uri else create_engine("sqlite://:memory:")
         self.engine = engine
-        self.Session = sessionmaker(bind=engine)
-        session = self.Session()
+        self.Session = scoped_session(sessionmaker(bind=engine))
+        session = self.Session(autocommit=True)
         metadata = MetaData(bind=engine) 
         self.sequence = Serial(engine, metadata, create=create, schema=schema)
         self.metadata = metadata
             for r in res:
                 if r.name != 'string':
                     self.registeType(r.name, storage=r.storage,)
-        session.close()
+
+    def __del__(self):
+        self.close()
+
+    def close(self):
+        self.Session.close()
 
     def registeType(self, typename, storage=None, colType=None, create=False, ):
         """
 
         subjects = self.registed_storage['subject']
         strings = self.registed_storage['string']
+        session = self.Session()
+        session.begin()
 
         for segmentType in self.registed_type.values():
             if storname == segmentType.table.name:
                 self.registed_type[typename] = segmentType
                 session.execute(subjects.insert().values(predicate_id=meta.IS, obj=meta.TYPE),)
                 session.execute(strings.insert().values(predicate_id=meta.STORAGE, obj=table.name),)
-                session.commit()
                 return segmentType
 
         if create:
-            session = self.Session()
             if colType == None:
                 raise 'must give the object storage type if you want to create storage'
 
             session.execute(subjects.insert().values(subject_id=typeid, predicate_id=meta.IS, obj=meta.TYPE),)
             session.execute(strings.insert().values(subject_id=typeid, predicate_id=meta.NAME, obj=typename),)
             session.execute(strings.insert().values(subject_id=typeid, predicate_id=meta.STORAGE, obj=table.name),)
-            session.commit()
-            session.close()
         else:
             table = Table(storname, self.metadata, 
                           Column('subject_id', Integer, default=self.sequence), 
         self.registed_storage[typename] = table
         segmentType = self.__make_type_helper(typename, table)
         self.registed_type[typename] = segmentType
+        session.commit()
         return segmentType
 
     def newPredicate(self, name, objtype_name):
 >>> print when.objType.name
 datetime
         """
-        session = self.Session(autocommit=True)
+        session = self.Session()
+        session.begin()
         Subject = self.registed_type["subject"]
         subjects = Subject.table
         String = self.registed_type["string"]
                                              String.predicate_id==meta.NAME)).as_scalar()==name).scalar()
         if not pid:
             pid = self.sequence()
-        objType = session.query(strings.c.subject_id).\
-            filter(strings.c.predicate_id==meta.NAME).\
-            filter(strings.c.obj==objtype_name).\
-            filter(select([subjects.c.obj], 
-                          and_(subjects.c.subject_id==strings.c.subject_id,
-                               subjects.c.predicate_id==meta.IS,)).as_scalar()==meta.TYPE).scalar()
+        if objtype_name == 'subject':
+            objType = meta.SUBJECT
+        else:
+            objType = session.query(strings.c.subject_id).\
+                filter(strings.c.predicate_id==meta.NAME).\
+                filter(strings.c.obj==objtype_name).\
+                filter(select([subjects.c.obj], 
+                              and_(subjects.c.subject_id==strings.c.subject_id,
+                                   subjects.c.predicate_id==meta.IS,)).as_scalar()==meta.TYPE).scalar()
         storid = session.query(subjects.c.obj).\
             filter(subjects.c.subject_id==objType).\
             filter(subjects.c.predicate_id==meta.STORAGE).scalar()
         session.execute(subjects.insert().values(subject_id=pid, predicate_id=meta.IS, obj=meta.PREDICATE))
         session.execute(subjects.insert().values(subject_id=pid, predicate_id=meta.OBJTYPE, obj=objType))
         session.execute(strings.insert().values(subject_id=pid, predicate_id=meta.NAME, obj=name))
-        session.close()
+        session.commit()
         return pid
 
     def writeSubject(self, subject_id, segments, **pairs):
             sid = subject_id
         else:
             sid = self.sequence()
+        session = self.Session()
+        session.begin()
         for k,v in segments.iteritems():
-            self.writeSegment(sid, k, v)
+            self.writeSegment(sid, k, v, session)
         for k,v in pairs.iteritems():
-            self.writeSegment(sid, k, v)
+            self.writeSegment(sid, k, v, session)
+        session.commit()
         return sid
 
-    def writeSegment(self, subject_id, predicate, obj):
+    def writeSegment(self, subject_id, predicate, obj, session=None):
         """
         此版本未解决同名谓词问题。仅适用于单一谓词应用,即每个谓词名仅对应一条谓词定义。
 >>> from sqlalchemy import DateTime
 >>> print segment.subject_id
 12
         """
-        session=self.Session()
+        needclose = True
+        if session:
+            _session = session
+            needclose = False
+        else:
+            _session=self.Session()
+        if needclose:
+            _session.begin()
         subjects = self.registed_storage["subject"]
         strings = self.registed_storage["string"]
         if subject_id:
         else:
             sid = self.sequence()
         pid = self.get_preidcate_id_by_name(predicate)
-        objType = session.query(subjects.c.obj).\
+        objType = _session.query(subjects.c.obj).\
             filter(subjects.c.subject_id==pid).\
             filter(subjects.c.predicate_id==meta.OBJTYPE).scalar() 
-        typename = session.query(strings.c.obj).\
+        typename = _session.query(strings.c.obj).\
             filter(strings.c.subject_id==objType).\
             filter(strings.c.predicate_id==meta.NAME).scalar()
         SegmentType = self.registed_type[typename]
         segment = SegmentType(sid, pid, obj)
-        session.add(segment)
-        session.commit()
-        session.close()
+        _session.add(segment)
+        if needclose:
+            _session.commit()
         return segment 
 
     def initStorage(self, schema=None):
         meta.init_meta(self.engine, self.registed_storage["subject"], 
                        self.registed_storage["string"], self.sequence, clearfirst)
 
-    def getSegmentsBySubjectId(self, subject_id):
-        session = self.Session(autocommit=True)
+    def getSegmentsBySubjectId(self, subject_id, session=None):
+        if session:
+            _session=session
+        else:
+            _session = self.Session()
         for Type in self.registed_type.values():
-            for segment in session.query(Type).filter(Type.subject_id==subject_id):
+            for segment in _session.query(Type).filter(Type.subject_id==subject_id):
                 yield segment
-        session.close()
+        if session == None:
+            _session.commit()
+        #session.close()
 
     def getSubjectById(self, subject_id):
         """
 >>> storage = Storage("sqlite://", create=True)
->>> pid = storage.newPredicate('kssoid', 'string')
->>> kssoid = storage.getSubjectById(pid)
->>> print kssoid.name
-kssoid
->>> segment = storage.writeSegment(500, 'kssoid', 'liuxin2')
+>>> pid = storage.newPredicate('ssoid', 'string')
+>>> print pid
+10
+>>> ssoid = storage.getSubjectById(pid)
+>>> print ssoid.name
+ssoid
+>>> segment = storage.writeSegment(500, 'ssoid', 'March Liu')
 >>> print segment.predicate.name
-kssoid
+ssoid
 >>> subject = storage.getSubjectById(segment.subject_id)
 >>> print subject.pnames
-set([u'kssoid'])
+set([u'ssoid'])
         """
-        segments = [segment for segment in self.getSegmentsBySubjectId(subject_id)]
+        session = self.Session()
+        segments = [segment for segment in self.getSegmentsBySubjectId(subject_id, session)]
         subject_is = [segment 
                       for segment in segments 
                       if segment.predicate_id==meta.IS and segment.obj in (meta.PREDICATE, meta.TYPE)]
         """
         SegmentString = self.registed_type['string']
         SegmentSubject = self.registed_type['subject']
-        session=self.Session(autocommit=True)
+        session=self.Session()
         re = session.query(SegmentSubject.subject_id,).\
             filter(SegmentSubject.predicate_id==meta.IS).\
             filter(SegmentSubject.obj==meta.PREDICATE).\
             filter(select([SegmentString.obj], 
                           and_(SegmentString.subject_id==SegmentSubject.subject_id,
                                SegmentString.predicate_id==meta.NAME)).as_scalar()==name).scalar()
+        #session.close()
         return re
 
     def __make_type_helper(self, typename, table):