query - Ejemplo de uso de BLOB en SQLAlchemy

sqlalchemy query

Question

¿Alguien tiene un ejemplo de cómo usar BLOB en SQLAlchemy?

Saludos, Steve

Answer 1

from sqlalchemy import *
from sqlalchemy.orm import mapper, sessionmaker
import os
engine = create_engine(''sqlite://'', echo=True)
metadata = MetaData(engine)
sample = Table(
''sample'', metadata,
Column(''id'', Integer, primary_key=True),
Column(''lob'', Binary),
)
class Sample(object):
def __init__(self, lob):
self.lob = lob
mapper(Sample, sample)
metadata.create_all()
session = sessionmaker(engine)()
# Creating new object
blob = os.urandom(100000)
obj = Sample(lob=blob)
session.add(obj)
session.commit()
obj_id = obj.id
session.expunge_all()
# Retrieving existing object
obj = session.query(Sample).get(obj_id)
assert obj.lob==blob

Answer 2

De la documentación BINARY parece el camino a seguir: http://docs.sqlalchemy.org/en/latest/dialects/mysql.html

clase sqlalchemy.dialects.mysql.BLOB (length = None) Bases: sqlalchemy.types.LargeBinary

El tipo SQL BLOB.

init (length = None) Construye un tipo LargeBinary.

Parámetros: length - opcional, una longitud para la columna para uso en sentencias DDL, para aquellos tipos BLOB que aceptan una longitud (es decir, MySQL). No produce un tipo BINARIO / VARBINARIO alargado - use los tipos BINARIO / VARBINARIO específicamente para aquellos. Se puede omitir con seguridad si no se emitirá CREATE TABLE. Ciertas bases de datos pueden requerir una longitud para su uso en DDL, y generarán una excepción cuando se emite CREATE TABLE DDL.

Answer 3

from sqlalchemy import *
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from struct import *
_DeclarativeBase = declarative_base()
class MyTable(_DeclarativeBase):
__tablename__ = ''mytable''
id = Column(Integer, Sequence(''my_table_id_seq''), primary_key=True)
my_blob = Column(BLOB)
DB_NAME = ''sqlite:///C:/BlobbingTest.db''
db = create_engine(DB_NAME)
#self.__db.echo = True
_DeclarativeBase.metadata.create_all(db)
Session = sessionmaker(bind=db)
session = Session()
session.add(MyTable(my_blob=pack(''H'', 365)))
l = [n + 1 for n in xrange(10)]
session.add(MyTable(my_blob=pack(''H''*len(l), *l)))
session.commit()
query = session.query(MyTable)
for mt in query.all():
print unpack(''H''*(len(mt.my_blob)/2), mt.my_blob)