import datetime
sl_starttime= datetime.datetime.utcnow()
import copy
import collections
import datetime
import enum
import json
import flask
import sqlalchemy
import sqlalchemy.ext.declarative
import sqlalchemy.orm
import sqlalchemy.orm.query
import sqlalchemy.sql
import sqlalchemy.types
import sys
import uuid
import warnings
from config import app, db_login, db_password, db_remote_adress,
db_remote_port, db_name, db_echo, db_isolation_level, ignore_list
engine =
sqlalchemy.create_engine('postgresql+psycopg2://'+db_login+':'+db_password+'@'+db_remote_adress+':'+str(db_remote_port)+'/'+db_name,
echo=db_echo, isolation_level=db_isolation_level)
engine.execute("SET TIME ZONE 0")
Base = sqlalchemy.ext.declarative.declarative_base()
session = sqlalchemy.orm.sessionmaker(bind = engine)()
class UUID(sqlalchemy.types.TypeDecorator):
impl = sqlalchemy.types.BINARY
def load_dialect_impl(self, dialect):
if dialect.name == 'postgresql':
return dialect.type_descriptor(sqlalchemy.dialects.postgresql.UUID())
else:
return dialect.type_descriptor(sqlalchemy.types.BINARY(16))
def process_bind_param(self, value, dialect):
if value is None:
return value
else:
if not isinstance(value, uuid.UUID):
if isinstance(value, bytes):
value = uuid.UUID(bytes=value)
elif isinstance(value, int):
value = uuid.UUID(int=value)
elif isinstance(value, str):
value = uuid.UUID(value)
if dialect.name == 'postgresql':
return str(value)
else:
return value.bytes
def process_result_value(self, value, dialect):
if value is None:
return value
if dialect.name == 'postgresql':
return uuid.UUID(value)
else:
return uuid.UUID(bytes=value)
engine.execute('CREATE EXTENSION IF NOT EXISTS "uuid-ossp";')
warnings.filterwarnings("ignore")
class Basic_class(Base):
__tablename__ = "documentsrelationship_"
all_id =sqlalchemy.Column(UUID,
server_default=sqlalchemy.text("uuid_generate_v4()"), primary_key=True)
polymorphic_type = sqlalchemy.Column(sqlalchemy.types.String)
__mapper_args__ =
{'polymorphic_identity':__tablename__,'polymorphic_on':polymorphic_type}
class db_factory_class ():
def __getattr__(self, attr):
if False:
pass
elif attr == "Class_a":
class Class_a(Base):
__tablename__ = "class_a"
all_id = sqlalchemy.Column(UUID,
sqlalchemy.ForeignKey(Basic_class.all_id), primary_key = True)
__mapper_args__ = {"polymorphic_identity" : __tablename__,
"inherit_condition" : all_id == Basic_class.all_id}
name = sqlalchemy.Column(sqlalchemy.types.String)
def __getattr__ (self2, key2):
if key2 == "_sa_instance_state":
return Base.__getattr__(self2, key2)
elif key2 == "Class_b":
load_function("Class_a", "Class_b")
else:
getattr(self.__class__.__bases__[0], key2)
def __setattr__ (self2, key2, value2):
if key2 == "_sa_instance_state":
Base.__setattr__(self2, key2, value2)
else:
old_value = getattr(self2, key2)
setattr(self2, key2, value2)
class_factory.Class_a = Class_a
elif attr == "Class_b":
class Class_b(Base):
__tablename__ = "class_b"
all_id = sqlalchemy.Column(UUID,
sqlalchemy.ForeignKey(Basic_class.all_id), primary_key = True)
__mapper_args__ = {"polymorphic_identity" : __tablename__,
"inherit_condition" : all_id == Basic_class.all_id}
name = sqlalchemy.Column(sqlalchemy.types.String)
def __getattr__ (self2, key2):
if key2 == "_sa_instance_state":
return Base.__getattr__(self2, key2)
elif key2 == "Class_a":
load_function("Class_b", "Class_a")
else:
getattr(self.__class__.__bases__[0], key2)
def __setattr__ (self2, key2, value2):
if key2 == "_sa_instance_state":
Base.__setattr__(self2, key2, value2)
else:
old_value = getattr(self2, key2)
setattr(self2, key2, value2)
class_factory.Class_b = Class_b
return getattr(self, attr)
class_factory = db_factory_class()
class_factory.Class_a()
def load_function (class_name, attr_name):
print ('called preload for ' + str(class_name) + ' attr '+ str(attr_name))
if False:
pass
elif class_name == "Class_b" and attr_name == "Class_a":
class_factory.Class_b.Class_a_id = sqlalchemy.Column(UUID, sqlalchemy.
ForeignKey(class_factory.Class_a.mRID), nullable = True)
class_factory.Class_b.Class_a =
sqlalchemy.orm.relationship(class_factory.Class_a, foreign_keys =
class_factory.Class_b.Class_a_id,\
uselist = False, backref = sqlalchemy.orm.backref("Class_b", uselist =
True))
elif class_name == "Class_a" and attr_name == "Class_b":
load_function("Class_b", "Class_a")
else:
raise ValueError ("invalid class/attribute")
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
