Hi all,
my apologies for a long question, I am new to sqlalchemy and I was
struggling with one problem. I solved it but I really not sure it's
the right way to do it.
The problem is that I need to create objects like this:
Token(pos=2, surface="helloWorld")
The surface is string, but in the database, it should be saved as
reference to an other table (looked-up and saved if not existing)
The way I solved it was to create my own column type decorating
Integer. On the way in, it gets text and stores id, on the way out, it
gives the text back.
But I don't know if my solution is safe. For instance,
-- I am decorating Integer type, but on the output there is string.
-- I am using the EntitySingleton recipy, but I found out I could not
use the recommended
Session = scoped_session(sessionmaker())
I had to say: Session = sessionmaker()
-- And I do commit for every new looked-up value
And also, I have the impression the solution is very complicated,
perhaps you could help me to do it in a better way.
Many thanks!
roman
Here is the code with an example:
-------------------------------------------------------
import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker, relation,
backref, mapper
from sqlalchemy import create_engine, Table, Column, Integer, String,
MetaData, ForeignKey
#from mimetypes import types_map
import weakref
import sqlalchemy.types as types
Session = sessionmaker() #scoped_session(sessionmaker)
DBase = declarative_base()
class MyUnique(types.TypeDecorator):
"""a type that decorates Integer,
receives a text representation - eg. MyValueFromEnumeration
and replaces it in the database with id of the string above
It makes sure that inserted values are unique (creates them if not
exist)"""
impl = types.Integer
def process_bind_param(self, value, dialect):
#ukladani do databaze
if not hasattr(self, '__unique_valuee'):
self.__unique_valuee = Surface(value)
return self.__unique_valuee.id
def process_result_value(self, value, dialect):
#vystup
return self.__unique_valuee.surface
""" #if it is string type
def copy(self):
return MyUniqueString(self.impl.length)
"""
class DocumentBase(object):
def __init__(self):
self.metadata = DBase.metadata
self.engine = None
def open(self, url='sqlite:///:memory:'):
self.engine = create_engine(url, echo=True)
#engine = create_engine('sqlite:///' + __file__ + ".db",
echo=True)
Session.configure(bind=self.engine)
self.metadata.bind = self.engine
surface_table = sa.Table('surface', self.metadata,
sa.Column('id', sa.Integer, primary_key=True,
autoincrement=True),
sa.Column('surface', sa.String(500), autoincrement=True)
)
mapper(Surface, surface_table)
self.metadata.create_all(checkfirst=True)
def add(self, document_object):
sess = Session()
sess.save_or_update(document_object)
sess.commit()
class EntitySingleton(type):
"""a metaclass that insures the creation of unique and non-
existent entities for
a particular constructor argument. if an entity with a particular
constructor
argument was already created, either in memory or in the database,
it is returned
in place of constructing the new instance."""
def __init__(cls, surface, bases, dct):
cls.instances = weakref.WeakValueDictionary()
def __call__(cls, surface):
session = Session()
hashkey = surface
try:
return cls.instances[hashkey]
except KeyError:
instance =
session.query(cls).filter(cls.surface==surface).first()
if instance is None:
instance = type.__call__(cls, surface)
#session.begin_nested()
session.add(instance)
session.commit()
# optional - flush the instance when it's saved
session.flush([instance])
cls.instances[hashkey] = instance
return instance
class Document(DBase):
__tablename__ = 'documents'
id = sa.Column(sa.Integer(9), primary_key=True,
autoincrement=True)
name = sa.Column(sa.String(500))
uri = sa.Column(sa.String(500))
processing_stage = sa.Column(sa.String(500))
tokens = relation("Token", backref="Document")
def __xinit__(self, name):
self.name = name
def __repr__(self):
return "<%s('%s','%s')>" % (self.__class__, self.id,
self.name)
class Token(DBase):
__tablename__ = 'tokens'
id = sa.Column(sa.Integer(6), primary_key=True,
autoincrement=True)
pos = sa.Column(sa.Integer(6))
type = sa.Column(sa.String(500))
surface = sa.Column(MyUnique(500), sa.ForeignKey('surface.id'))
surface_translated = sa.Column(sa.String(500))
sem = sa.Column(sa.String(500))
title_id = sa.Column(sa.Integer, sa.ForeignKey('documents.id'))
#document = relation("Document", backref="Token")
#def __init__(self, surface=None, **kwargs):
def __repr__(self):
return "<%s('%s','%s')>" % (self.__class__, self.id,
self.surface)
class Surface(object):
__metaclass__ = EntitySingleton
#__tablename__ = 'surface'
#id = sa.Column(sa.Integer(6), primary_key=True,
autoincrement=True)
#surface = sa.Column(sa.String(500), primary_key=True)
def __init__(self, value):
self.surface = value
def test_database():
session = Session()
doc = Document(uri="xx1", name="TestDoc")
doc.name = "Test"
for i in range(1,5):
t = Token(pos=i, surface="Text" + str(i))
doc.tokens.append(t)
#insert the same values again
for i in range(1,5):
t = Token(pos=i, surface="Text" + str(i))
doc.tokens.append(t)
session.add(doc)
session.commit()
print doc.tokens
print
if __name__ == "__main__":
db = DocumentBase()
db.open()
test_database()
--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to sqlalchemy@googlegroups.com
To unsubscribe from this group, send email to [EMAIL PROTECTED]
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en
-~----------~----~----~----~------~----~------~--~---
