Sorry, the code was too big, so I leave here the file.
--
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 sqlalchemy+unsubscr...@googlegroups.com.
To view this discussion on the web visit
https://groups.google.com/d/msgid/sqlalchemy/163a8327-1d16-43bf-baa0-a327aa6e2b18%40googlegroups.com.
import uuid
import json
import datetime
from src.common.helpers import format_date_TZ
from sqlalchemy import create_engine
from sqlalchemy.orm import relationship, create_session
from sqlalchemy import (
Column, DateTime, String, Integer,
Boolean, ForeignKey, Table
)
from src.application.kernel.engine import Base
STRING_ID = 50
STRING_JSON = 1024
STRING_NAME = 100
STRING_TYPE = 10
STRING_BIG_JSON = 2048
class Page(Base):
__tablename__ = 'page'
# Columns
page_uuid = Column(String(STRING_ID), nullable=False)
id = Column(
String(STRING_ID),
primary_key=True,
nullable=False
)
event_date = Column(DateTime)
last_update = Column(
DateTime,
default=datetime.datetime.now(),
onupdate=datetime.datetime.now(),
)
creation_date = Column(DateTime(timezone=True))
title = Column(String(STRING_NAME), nullable=False)
slug = Column(String(STRING_NAME), nullable=False)
description = Column(String(STRING_BIG_JSON))
perks = Column(String(STRING_BIG_JSON))
meta_description = Column(String(STRING_JSON))
about = Column(String(STRING_BIG_JSON))
status = Column(String(STRING_TYPE), nullable=False)
keywords = Column(String(STRING_JSON))
seo_title = Column(String(STRING_JSON))
search_field = Column(String(STRING_JSON))
marketing_type = Column(String(STRING_JSON))
images = Column(String(STRING_JSON))
cover_images = Column(String(STRING_JSON))
graph_images = Column(String(STRING_JSON))
thumbnails = Column(String(STRING_JSON))
hide_from_search = Column(Boolean)
page_type = Column(String(STRING_JSON), nullable=False)
__mapper_args__ = {
'polymorphic_identity': 'page',
'polymorphic_on': page_type,
'with_polymorphic': '*',
# This triggers the update on cascade
# https://docs.sqlalchemy.org/en/13/orm/mapping_api.html?highlight=passive_updates#sqlalchemy.orm.mapper.params.passive_updates
'passive_updates': False,
}
def __init__(
self,
data=None,
page_type=None,
images=None,
c_images=None,
g_images=None,
thumbnails=None,
tags=None,
ass_prod=None,
):
print('### INIT PAGE')
self.page_uuid = str(uuid.uuid4())
if ass_prod:
self.update(data, tags=tags, ass_prod=ass_prod)
self.update(data, tags=tags)
def update(
self,
data=None,
page_type=None,
images=None,
c_images=None,
g_images=None,
thumbnails=None,
):
print('**UPDATING PAGE')
if data:
self.id = data.get('id').get('__value')
if data.get('event_date'):
self.event_date = data.get('event_date').get('__value')
self.creation_date = format_date_TZ(
data.get('createdAt').get('__value')
) if data.get('createdAt').get('__value') else None
self.title = data.get('title').get('__value')
self.slug = data.get('slug').get('__value')
self.description = data.get('description').get('__value')
self.perks = json.dumps(data.get('perks').get('__value'))
self.meta_description = json.dumps(data.get('metaDescription').get('__value'))
self.about = data.get('about').get('__value')
self.status = data.get('status').get('__value').get('status')
self.keywords = json.dumps(data.get('keywords').get('__value'))
self.seo_title = data.get('SEOTitle').get('__value')
self.marketing_type = data.get('type').get('__value')
self.search_field = ''
self.hide_from_search = True
if page_type:
self.page_type = page_type
if images:
self.images = images
if c_images:
self.cover_images = c_images
if g_images:
self.graph_images = g_images
if thumbnails:
self.thumbnails = thumbnails
product_tag_association = Table(
'product_tag_association',
Base.metadata,
Column(
'p_id',
String(STRING_ID),
ForeignKey('product.id'),
primary_key=True
),
Column(
't_id',
String(STRING_ID),
ForeignKey('pagetag.tag_id'),
primary_key=True
)
)
class Product(Page):
__tablename__ = 'product'
# Columns
product_uuid = Column(String(STRING_ID), nullable=False)
id = Column(
String(STRING_ID),
ForeignKey('page.id'),
primary_key=True,
nullable=False
)
concessions = Column(String(STRING_BIG_JSON), nullable=False)
product_type = Column(String(STRING_TYPE), nullable=False)
tags = relationship(
"Tag",
secondary="product_tag_association",
backref="products"
)
__mapper_args__ = {
'polymorphic_identity': 'product',
# 'polymorphic_on': product_type,
}
def __init__(self, data=None, product_type=None, tags=None, ass_prod=None):
print('### INIT PRODUCT')
super().__init__(
data=data,
page_type='product',
tags=tags,
ass_prod=ass_prod,
)
self.product_uuid = str(uuid.uuid4())
def update(self, data=None, product_type=None, tags=None):
print('**UPDATING PRODUCT')
print(tags)
# Here I had my main error. I was not passing down the product_type (which in deed will be a page_type)
# I was passing down page_type='product' so when I generate an entity son of Product, its page_type
# was overwritten by product.
super().update(data=data, page_type=product_type)
if data:
# self.id = data.get('id').get('__value')
self.concessions = json.dumps(data.get('concessions').get('__value'))
if product_type:
self.product_type = product_type
if tags:
self.tags = tags
class Tag(Page):
__tablename__ = 'pagetag'
# Columns
tag_uuid = Column(String(STRING_ID), nullable=False)
tag_id = Column(
String(STRING_ID),
ForeignKey('page.id'),
primary_key=True,
nullable=False
)
featured = Column(Integer, nullable=False)
featured_index = Column(Integer, default=0, nullable=False)
tag_type = Column(String(STRING_TYPE), nullable=False)
__mapper_args__ = {
'polymorphic_identity': 'pagetag',
}
def __init__(self, data=None):
print('### INIT TAG')
super().__init__(data, page_type='pagetag')
self.tag_uuid = str(uuid.uuid4())
def update(self, data=None, tags=None):
super().update(data=data, page_type='pagetag')
if data:
self.featured = 0
self.featured_index = 0
self.tag_type = data.get('tagType').get('__value')
travelpass_travel_association = Table(
'travelpass_travel_association',
Base.metadata,
Column(
'tp_id',
String(STRING_ID),
ForeignKey('travel_pass.id'),
primary_key=True
),
Column(
't_id',
String(STRING_ID),
ForeignKey('travel.id'),
primary_key=True
)
)
class TravelPass(Product):
__tablename__ = 'travel_pass'
# Columns
travel_pass_uuid = Column(String(STRING_ID), nullable=False)
id = Column(
String(STRING_ID),
ForeignKey('product.id'),
primary_key=True,
nullable=False
)
allow_one_way_optin = Column(Boolean, default=False)
auto_journey_optin = Column(Boolean, default=False)
allow_repurchase = Column(Boolean, default=False)
allow_edit_booking = Column(Boolean, default=False)
allow_guests_onboard = Column(Boolean, default=False)
create_own_page = Column(Boolean, default=False)
show_departure_time = Column(Boolean, default=False)
opt_out_promo = Column(String(STRING_JSON))
max_number_passenger = Column(Integer)
start_date = Column(DateTime(timezone=True))
end_when = Column(String(STRING_JSON))
end_date = Column(DateTime(timezone=True))
end_after_amount = Column(Integer)
end_after_unit = Column(String(STRING_JSON))
allow_return = Column(Boolean, default=False)
# Columns with relationships
associated_products = relationship(
'Travel',
secondary=travelpass_travel_association,
back_populates='associated_travelpass'
)
__mapper_args__ = {
'polymorphic_identity': 'travel_pass',
}
def __init__(self, data=None, ass_prod=None, tags=None):
print('### INIT TRAVELPASS')
super().__init__(data=data, tags=tags, ass_prod=ass_prod)
self.travel_pass_uuid = str(uuid.uuid4())
def update(self, data=None, travel_type=None, tags=None, ass_prod=None):
print('**UPDATING TRAVEL PASS')
super().update(data=data, product_type='travel_pass', tags=tags)
if data:
# super().__init__(data=data, product_type='travel_pass')
self.id = data.get('id').get('__value')
self.allow_one_way_optin = (
data
.get('allowOneWayOptin')
.get('__value')
)
self.auto_journey_optin = (
data
.get('autoJourneyOptin')
.get('__value')
)
self.allow_repurchase = data.get('allowRepurchase').get('__value')
self.allow_edit_booking = (
data
.get('allowEditBooking')
.get('__value')
)
self.allow_guests_onboard = (
data
.get('allowGuestsOnBoard')
.get('__value')
)
self.create_own_page = data.get('createOwnPage').get('__value')
self.show_departure_time = (
data
.get('showDepartureTime')
.get('__value')
)
self.opt_out_promo = json.dumps(
data.get('optOutPromo').get('__value')
)
self.max_number_passengers = (
data
.get('maxNumberPassengers')
.get('__value')
)
self.start_date = format_date_TZ(
data.get('startDate').get('__value')
)
self.end_when = (
data
.get('endWhen')
.get('__value')
.get('end_when')
) if data.get('endWhen').get('__value') else None
self.end_date = format_date_TZ(
data.get('endDate').get('__value')
) if data.get('endDate').get('__value') else None
self.end_after_amount = data.get('endAfterAmount').get('__value')
self.end_after_unit = (
data
.get('endAfterUnit')
.get('__value')
.get('end_after_unit')
) if data.get('endAfterUnit').get('__value') else None
self.allow_return = data.get('allowReturn').get('__value')
if ass_prod:
self.associated_products = ass_prod
class Travel(Product):
__tablename__ = 'travel'
# Columns
travel_uuid = Column(String(STRING_ID), nullable=False)
id = Column(
String(STRING_ID),
ForeignKey('product.id'),
primary_key=True,
nullable=False
)
outbound_seats_on_sale = Column(Integer, default=0)
return_seats_on_sale = Column(Integer, default=0)
travel_type = Column(String(STRING_TYPE), nullable=False)
# Columns with relationships
associated_travelpass = relationship(
'TravelPass',
secondary=travelpass_travel_association,
back_populates='associated_products'
)
__mapper_args__ = {
'polymorphic_identity': 'travel',
# 'polymorphic_on': travel_type
}
def __init__(self, data=None, travel_type=None, tags=None):
print('### INIT TRAVEL')
super().__init__(data=data, product_type='travel', tags=tags)
self.travel_uuid = str(uuid.uuid4())
def update(self, data=None, travel_type=None, tags=None, ass_prod=None):
print('**UPDATING TRAVEL')
super().update(data=data, product_type='travel', tags=tags)
if data:
# self.id = data.get('id').get('__value')
self.outbound_seats_on_sale = (
data
.get('outboundSeatsOnSale')
.get('__value')
)
self.return_seats_on_sale = (
data
.get('returnSeatsOnSale')
.get('__value')
)
if travel_type:
self.travel_type = travel_type
class Event(Travel):
__tablename__ = 'event'
# Columns
event_uuid = Column(String(STRING_ID), nullable=False)
id = Column(
String(STRING_ID),
ForeignKey('travel.id'),
primary_key=True,
nullable=False
)
from_date = Column(DateTime(timezone=True), nullable=False)
to_date = Column(DateTime(timezone=True), nullable=False)
__mapper_args__ = {
'polymorphic_identity': 'event',
}
def __init__(self, data=None, tags=None):
print('### INIT EVENT')
super().__init__(data=data, travel_type='event', tags=tags)
self.event_uuid = str(uuid.uuid4())
# self.update(data)
def update(self, data=None, tags=None, ass_prod=None):
super().update(data=data, travel_type='event', tags=tags)
if data:
# self.id = data.get('id').get('__value')
self.from_date = format_date_TZ(
data.get('eventFromDate').get('__value')
) if data.get('eventFromDate').get('__value') else None
self.to_date = format_date_TZ(
data.get('eventToDate').get('__value')
) if data.get('eventToDate').get('__value') else None
e = create_engine("sqlite:///test.db", echo=True)
Base.metadata.drop_all(e)
Base.metadata.create_all(e)
travelpass_data = {
'old_entity': None,
'new_entity': {
'__type': 'array',
'__encrypted': False,
'__class': None,
'__value': {
'id': {
'__value': '4753fa6a-9b9f-47b9-975c-dacb7634f093'
},
'createdAt': {
'__value': None
},
'title': {
'__value': 'Travel pass test title'
},
'slug': {
'__value': 'rides%2Fsporting-en-primera%2Ftravel-pass-test-title'
},
'description': {
'__value': 'Travel pass test description'
},
'perks': {
'__value': []
},
'images': {
'__value': []
},
'coverImages': {
'__value': []
},
'metaDescription': {
'__value': 'Travel pass test description SEO BODY'
},
'about': {
'__value': '<p><strong>Travel pass about information</strong></p>\n'
},
'status': {
'__value': {'status': 'UNPUBLISHED'}
},
'keywords': {
'__value': 'travel'
},
'graphImages': {
'__value': []
},
'SEOTitle': {
'__value': 'Travel pass test title SEO TITLE'
},
'thumbnails': {
'__value': []
},
'type': {
'__value': 'sport'
},
'tags': {
'__value': [
{
'__value': '27422bce-dd24-45fa-9096-f402c1f2c9a5'
},
{
'__value': '2fc57cf6-3022-4926-b597-5798d259d693'
}
]
},
'concessions': {
'__value': []
},
'allowOneWayOptin': {
'__value': True
},
'autoJourneyOptin': {
'__value': True
},
'allowRepurchase': {
'__value': True
},
'allowEditBooking': {
'__value': True
},
'allowGuestsOnBoard': {
'__value': True
},
'createOwnPage': {
'__value': False
},
'showDepartureTime': {
'__value': True
},
'optOutPromo': {
'__value': None
},
'maxNumberPassengers': {
'__value': 100
},
'associateProducts': {
'__value': [
{
'__value': '21bd039b-465b-4de3-bf40-fc1b4a41843e'
},
{
'__value': '553acea9-22d1-4118-954a-7bc699166fcc'
}
]
},
'startDate': {
'__value': '20190514140000 Europe/London'
},
'endWhen': {
'__value': {
'end_when': 'ON_SPECIFIC_DATE'
}
},
'endDate': {
'__value': '20190530005700 Europe/London'
},
'endAfterAmount': {
'__value': None
},
'endAfterUnit': {
'__value': None
},
'allowReturn': {
'__value': True
}
},
'__entity_version': 1
}
}
session = create_session(bind=e, autoflush=True, autocommit=False)
data = travelpass_data.get('new_entity').get('__value')
print(data)
ass_prod = []
for prod in data.get('associateProducts').get('__value'):
temp = (
session
.query(Travel)
.filter(Travel.id == prod.get('__value'))
.all()
)
if temp:
ass_prod.append(temp)
ass_tags = []
for tag_id in data.get('tags').get('__value'):
ass_tags.append(
session.query(Tag)
.filter(
Tag.id == tag_id.get('__value')
)
.one_or_none()
)
print(ass_tags)
new_entity = TravelPass(
data=data,
ass_prod=ass_prod,
tags=None
)
session.add(new_entity)
session.commit()
new_entry_id = data.get('id').get('__value')
print(TravelPass)
print('REQUIRED ID: {}'.format(new_entry_id))
entry_exist = (
session
.query(TravelPass)
.filter(TravelPass.id == new_entry_id)
.one_or_none()
)
print('****ENTITY TYPE: {}'.format(entry_exist.__class__))
print(dict(entry_exist.__dict__))
