270 lines
9.7 KiB
Python
270 lines
9.7 KiB
Python
"""devices1
|
|
|
|
Revision ID: 34fe3d061c5f
|
|
Revises: 8c06ac60d125
|
|
Create Date: 2020-10-27 19:38:48.445908
|
|
|
|
"""
|
|
from datetime import date, timedelta
|
|
|
|
import sqlalchemy as sa
|
|
|
|
from alembic import op
|
|
|
|
# revision identifiers, used by Alembic.
|
|
from sqlalchemy import column, func, select, table, text
|
|
from sqlalchemy.dialects import postgresql
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
|
|
|
|
revision = "34fe3d061c5f"
|
|
down_revision = "8c06ac60d125"
|
|
branch_labels = None
|
|
depends_on = None
|
|
|
|
|
|
def update_sale_categories():
|
|
product = table(
|
|
"products",
|
|
column("product_id", postgresql.UUID(as_uuid=True)),
|
|
column("sale_category_id", postgresql.UUID(as_uuid=True)),
|
|
column("sale_category_name", sa.Unicode(length=255)),
|
|
)
|
|
sale_category = table(
|
|
"sale_categories", column("id", postgresql.UUID(as_uuid=True)), column("name", sa.Unicode(length=255))
|
|
)
|
|
op.execute(
|
|
product.update(
|
|
values={
|
|
"sale_category_id": select([sale_category.c.id]).where(
|
|
product.c.sale_category_name == sale_category.c.name
|
|
)
|
|
}
|
|
)
|
|
)
|
|
with op.batch_alter_table("products") as batch_op:
|
|
batch_op.alter_column("sale_category_id", nullable=False)
|
|
batch_op.drop_column("sale_category_name")
|
|
|
|
|
|
def update_section_printers():
|
|
section_printer = table(
|
|
"section_printers",
|
|
column("section_id", postgresql.UUID(as_uuid=True)),
|
|
column("section_name", sa.Unicode(length=255)),
|
|
column("printer_id", postgresql.UUID(as_uuid=True)),
|
|
column("printer_name", sa.Unicode(length=255)),
|
|
)
|
|
section = table("sections", column("id", postgresql.UUID(as_uuid=True)), column("name", sa.Unicode(length=255)))
|
|
printer = table(
|
|
"printers",
|
|
column("id", postgresql.UUID(as_uuid=True)),
|
|
column("name", sa.Unicode(length=255)),
|
|
column("cut_code", sa.Unicode(length=255)),
|
|
)
|
|
op.execute(
|
|
section_printer.update(
|
|
values={"section_id": select([section.c.id]).where(section_printer.c.section_name == section.c.name)}
|
|
)
|
|
)
|
|
op.execute(
|
|
section_printer.update(
|
|
values={"printer_id": select([printer.c.id]).where(section_printer.c.printer_name == printer.c.name)}
|
|
)
|
|
)
|
|
op.execute(printer.update(values={"cut_code": "\x1d\x56\x41\x03"}))
|
|
with op.batch_alter_table("section_printers") as batch_op:
|
|
batch_op.alter_column("section_id", nullable=False)
|
|
batch_op.drop_column("section_name")
|
|
batch_op.alter_column("printer_id", nullable=False)
|
|
batch_op.drop_column("printer_name")
|
|
|
|
|
|
def add_all_roles_to_owner():
|
|
r = table(
|
|
"roles",
|
|
column("id", postgresql.UUID(as_uuid=True)),
|
|
column("name", sa.Unicode(length=255)),
|
|
)
|
|
p = table(
|
|
"permissions",
|
|
column("id", postgresql.UUID(as_uuid=True)),
|
|
column("name", sa.Unicode(length=255)),
|
|
)
|
|
rp = table(
|
|
"role_permissions",
|
|
column("id", postgresql.UUID(as_uuid=True)),
|
|
column("permission_id", postgresql.UUID(as_uuid=True)),
|
|
column("role_id", postgresql.UUID(as_uuid=True)),
|
|
)
|
|
op.execute(
|
|
pg_insert(rp)
|
|
.from_select(
|
|
[rp.c.role_id, rp.c.permission_id],
|
|
select([select([func.distinct(r.c.id)]).where(r.c.name.ilike("Owner")).as_scalar(), p.c.id]),
|
|
)
|
|
.on_conflict_do_nothing(),
|
|
)
|
|
|
|
|
|
def rename_permissions():
|
|
p = table(
|
|
"permissions",
|
|
column("id", postgresql.UUID(as_uuid=True)),
|
|
column("name", sa.Unicode(length=255)),
|
|
)
|
|
rp = table(
|
|
"role_permissions",
|
|
column("id", postgresql.UUID(as_uuid=True)),
|
|
column("permission_id", postgresql.UUID(as_uuid=True)),
|
|
column("role_id", postgresql.UUID(as_uuid=True)),
|
|
)
|
|
op.execute(p.update().where(p.c.name == "Accounts Audit").values(name="Audit"))
|
|
op.execute(p.update().where(p.c.name == "Beer Consumption").values(name="Beer Sale Report"))
|
|
op.execute(p.update().where(p.c.name == "Bill Details").values(name="Bill Settlement Report"))
|
|
op.execute(p.update().where(p.c.name == "Cashier Checkout").values(name="Cashier Report"))
|
|
op.execute(p.update().where(p.c.name == "Sales Detail").values(name="Product Sale Report"))
|
|
op.execute(p.update().where(p.c.name == "Sales Analysis").values(name="Sale Report"))
|
|
op.execute(p.update().where(p.c.name == "Tax Analysis").values(name="Tax Report"))
|
|
op.execute(p.update().where(p.c.name == "Machines").values(name="Devices"))
|
|
op.execute(p.update().where(p.c.name == "Tables").values(name="Sections"))
|
|
|
|
op.execute(
|
|
rp.delete().where(
|
|
rp.c.permission_id == select([p.c.id]).where(p.c.name.ilike("Void or Reprinted Bill Report")).as_scalar()
|
|
)
|
|
)
|
|
op.execute(rp.delete().where(rp.c.permission_id == select([p.c.id]).where(p.c.name.ilike("Roles")).as_scalar()))
|
|
op.execute(
|
|
rp.delete().where(rp.c.permission_id == select([p.c.id]).where(p.c.name.ilike("NC Product")).as_scalar())
|
|
)
|
|
op.execute(
|
|
rp.delete().where(rp.c.permission_id == select([p.c.id]).where(p.c.name.ilike("Change Rate")).as_scalar())
|
|
)
|
|
op.execute(p.delete().where(p.c.name == "Void or Reprinted Bill Report"))
|
|
op.execute(p.delete().where(p.c.name == "Roles"))
|
|
op.execute(p.delete().where(p.c.name == "NC Product"))
|
|
op.execute(p.delete().where(p.c.name == "Change Rate"))
|
|
|
|
|
|
def update_settlements():
|
|
s = table(
|
|
"settlements",
|
|
column("voucher_id", postgresql.UUID(as_uuid=True)),
|
|
column("settled", sa.Integer()),
|
|
)
|
|
v = table(
|
|
"vouchers",
|
|
sa.Column("id", postgresql.UUID(as_uuid=True)),
|
|
sa.Column("voucher_type", sa.Integer()),
|
|
)
|
|
op.execute(
|
|
s.update()
|
|
.values(settled=9)
|
|
.where(s.c.voucher_id == v.c.id)
|
|
.where(v.c.voucher_type == 5)
|
|
.where(s.c.settled == 1)
|
|
)
|
|
|
|
|
|
def upgrade():
|
|
# ### commands auto generated by Alembic - please adjust! ###
|
|
update_sale_categories()
|
|
update_section_printers()
|
|
|
|
# Migration to temporal products
|
|
op.drop_constraint("fk_inventories_product_id_products", "inventories", type_="foreignkey")
|
|
op.drop_constraint(
|
|
"fk_modifier_categories_products_product_id_products", "modifier_categories_products", type_="foreignkey"
|
|
)
|
|
op.drop_constraint("uq_products_name", "products", type_="unique")
|
|
op.drop_constraint("pk_products", "products", type_="primary")
|
|
op.rename_table("products", "product_versions")
|
|
with op.batch_alter_table("product_versions") as batch_op:
|
|
batch_op.add_column(sa.Column("product_id", postgresql.UUID(as_uuid=True), nullable=True)),
|
|
batch_op.add_column(sa.Column("valid_from", sa.Date(), nullable=True))
|
|
batch_op.add_column(sa.Column("valid_till", sa.Date(), nullable=True))
|
|
batch_op.create_primary_key("pk_product_versions", ["id"])
|
|
|
|
p = table(
|
|
"product_versions",
|
|
column("is_active", sa.Boolean()),
|
|
column("valid_till", sa.Date()),
|
|
)
|
|
op.execute(
|
|
p.update().values(valid_till=date.today() - timedelta(days=1)).where(p.c.is_active == False) # noqa E712
|
|
)
|
|
|
|
op.drop_column("product_versions", "is_active")
|
|
|
|
prod_v = table(
|
|
"product_versions",
|
|
column("id", postgresql.UUID(as_uuid=True)),
|
|
column("product_id", postgresql.UUID(as_uuid=True)),
|
|
)
|
|
|
|
op.execute(prod_v.update(values={"product_id": prod_v.c.id}))
|
|
op.create_table(
|
|
"products",
|
|
sa.Column("id", postgresql.UUID(as_uuid=True), server_default=sa.text("gen_random_uuid()"), nullable=False),
|
|
sa.PrimaryKeyConstraint("id", name=op.f("pk_products")),
|
|
)
|
|
op.alter_column("product_versions", "product_id", nullable=False)
|
|
|
|
prod = table(
|
|
"product_versions",
|
|
column("id", postgresql.UUID(as_uuid=True)),
|
|
column("name", sa.Unicode(length=255)),
|
|
column("units", sa.Unicode(length=255)),
|
|
column("valid_from", sa.Date()),
|
|
column("valid_till", sa.Date()),
|
|
)
|
|
|
|
pv = table(
|
|
"products",
|
|
column("id", postgresql.UUID(as_uuid=True)),
|
|
)
|
|
|
|
op.execute(pv.insert().from_select([pv.c.id], select([prod.c.id])))
|
|
|
|
op.create_foreign_key(
|
|
op.f("fk_product_versions_product_id_products"), "product_versions", "products", ["product_id"], ["id"]
|
|
)
|
|
|
|
op.create_foreign_key(op.f("fk_inventories_product_id_products"), "inventories", "products", ["product_id"], ["id"])
|
|
op.create_foreign_key(
|
|
op.f("fk_modifier_categories_products_product_id_products"),
|
|
"modifier_categories_products",
|
|
"products",
|
|
["product_id"],
|
|
["id"],
|
|
)
|
|
|
|
op.alter_column("modifier_categories_products", "modifier_categories_id", new_column_name="modifier_category_id")
|
|
op.drop_constraint("uq_modifier_categories_products_product_id", "modifier_categories_products", type_="unique")
|
|
op.create_unique_constraint(
|
|
op.f("uq_modifier_categories_products_product_id"),
|
|
"modifier_categories_products",
|
|
["product_id", "modifier_category_id"],
|
|
)
|
|
|
|
op.create_exclude_constraint(
|
|
"uq_product_versions_name",
|
|
"product_versions",
|
|
(prod.c.name, "="),
|
|
(prod.c.units, "="),
|
|
(func.daterange(prod.c.valid_from, prod.c.valid_till, text("'[]'")), "&&"),
|
|
)
|
|
rename_permissions()
|
|
add_all_roles_to_owner()
|
|
update_settlements()
|
|
# ### end Alembic commands ###
|
|
|
|
|
|
def downgrade():
|
|
# ### commands auto generated by Alembic - please adjust! ###
|
|
op.add_column("products", sa.Column("sale_category_name", sa.Unicode(length=255), nullable=False))
|
|
op.drop_constraint("fk_products_sale_category_id_sale_categories", "products", type_="foreignkey")
|
|
op.drop_column("products", "sale_category_id")
|
|
# ### end Alembic commands ###
|