v6ole/H3ConuMS-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
dcb2435514762bf47f2eab423def23b761ab4142
H3ConuMS-v2/backend/alembic/versions/c3d4e5f6a7b8_add_inventory_tables.py
T
v6ole f1f8518985 ``` 
feat(auth): 添加用户权限获取接口并完善JWT令牌角色信息

- 在JWT令牌中添加用户角色信息
- 新增get_my_permissions接口用于获取当前用户权限码列表
- 重构认证回调逻辑，增加错误日志记录
- 更新用户信息获取接口使用Authorization头验证
```
2026-04-06 00:40:08 +08:00

193 lines
9.6 KiB
Python
Raw Blame History

"""add inventory management tables and permissions
Revision ID: c3d4e5f6a7b8
Revises: b2c3d4e5f6a7
Create Date: 2026-04-05
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
revision: str = 'c3d4e5f6a7b8'
down_revision: Union[str, None] = 'b2c3d4e5f6a7'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# 物料分类表
op.create_table(
'material_categories',
sa.Column('id', sa.BigInteger(), autoincrement=True, nullable=False),
sa.Column('name', sa.String(length=100), nullable=False),
sa.Column('code', sa.String(length=50), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('created_at', postgresql.TIMESTAMP(), server_default=sa.text('now()'), nullable=True),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('code'),
)
# 物料主数据表
op.create_table(
'materials',
sa.Column('id', sa.BigInteger(), autoincrement=True, nullable=False),
sa.Column('category_id', sa.BigInteger(), nullable=True),
sa.Column('name', sa.String(length=200), nullable=False),
sa.Column('model', sa.String(length=100), nullable=True),
sa.Column('specification', sa.Text(), nullable=True),
sa.Column('brand', sa.String(length=100), nullable=True),
sa.Column('unit', sa.String(length=20), server_default='个', nullable=True),
sa.Column('safe_quantity', sa.Integer(), server_default='0', nullable=True),
sa.Column('notes', sa.Text(), nullable=True),
sa.Column('created_at', postgresql.TIMESTAMP(), server_default=sa.text('now()'), nullable=True),
sa.Column('updated_at', postgresql.TIMESTAMP(), server_default=sa.text('now()'), nullable=True),
sa.ForeignKeyConstraint(['category_id'], ['material_categories.id']),
sa.PrimaryKeyConstraint('id'),
)
op.create_index('ix_materials_id', 'materials', ['id'], unique=False)
# 库存批次表
op.create_table(
'inventory_batches',
sa.Column('id', sa.BigInteger(), autoincrement=True, nullable=False),
sa.Column('material_id', sa.BigInteger(), nullable=True),
sa.Column('batch_no', sa.String(length=50), nullable=False),
sa.Column('quantity', sa.Integer(), nullable=False),
sa.Column('available_quantity', sa.Integer(), nullable=False, server_default='0'),
sa.Column('supplier', sa.String(length=200), nullable=True),
sa.Column('purchase_date', sa.Date(), nullable=True),
sa.Column('purchase_price', sa.Numeric(10, 2), nullable=True),
sa.Column('expiry_date', sa.Date(), nullable=True),
sa.Column('location', sa.String(length=100), nullable=True),
sa.Column('status', sa.String(length=20), server_default='in_stock', nullable=True),
sa.Column('created_at', postgresql.TIMESTAMP(), server_default=sa.text('now()'), nullable=True),
sa.ForeignKeyConstraint(['material_id'], ['materials.id']),
sa.PrimaryKeyConstraint('id'),
)
# 序列号设备表
op.create_table(
'serial_devices',
sa.Column('id', sa.BigInteger(), autoincrement=True, nullable=False),
sa.Column('material_id', sa.BigInteger(), nullable=True),
sa.Column('batch_id', sa.BigInteger(), nullable=True),
sa.Column('serial_no', sa.String(length=100), nullable=False),
sa.Column('mac_address', sa.String(length=17), nullable=True),
sa.Column('asset_no', sa.String(length=50), nullable=True),
sa.Column('status', sa.String(length=20), server_default='in_stock', nullable=True),
sa.Column('current_location', sa.String(length=200), nullable=True),
sa.Column('installed_info', postgresql.JSONB(), nullable=True),
sa.Column('onu_device_id', sa.BigInteger(), nullable=True),
sa.Column('notes', sa.Text(), nullable=True),
sa.Column('created_at', postgresql.TIMESTAMP(), server_default=sa.text('now()'), nullable=True),
sa.Column('updated_at', postgresql.TIMESTAMP(), server_default=sa.text('now()'), nullable=True),
sa.ForeignKeyConstraint(['batch_id'], ['inventory_batches.id']),
sa.ForeignKeyConstraint(['material_id'], ['materials.id']),
sa.ForeignKeyConstraint(['onu_device_id'], ['onu_devices.id']),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('serial_no'),
)
op.create_index('ix_serial_devices_mac_address', 'serial_devices', ['mac_address'], unique=False)
op.create_index('ix_serial_devices_status', 'serial_devices', ['status'], unique=False)
# 出入库记录表
op.create_table(
'inventory_transactions',
sa.Column('id', sa.BigInteger(), autoincrement=True, nullable=False),
sa.Column('transaction_no', sa.String(length=50), nullable=False),
sa.Column('transaction_type', sa.String(length=20), nullable=False),
sa.Column('material_id', sa.BigInteger(), nullable=True),
sa.Column('batch_id', sa.BigInteger(), nullable=True),
sa.Column('serial_device_id', sa.BigInteger(), nullable=True),
sa.Column('quantity', sa.Integer(), nullable=False),
sa.Column('from_status', sa.String(length=20), nullable=True),
sa.Column('to_status', sa.String(length=20), nullable=True),
sa.Column('operator_id', sa.BigInteger(), nullable=True),
sa.Column('project_name', sa.String(length=200), nullable=True),
sa.Column('installation_info', postgresql.JSONB(), nullable=True),
sa.Column('notes', sa.Text(), nullable=True),
sa.Column('created_at', postgresql.TIMESTAMP(), server_default=sa.text('now()'), nullable=True),
sa.ForeignKeyConstraint(['batch_id'], ['inventory_batches.id']),
sa.ForeignKeyConstraint(['material_id'], ['materials.id']),
sa.ForeignKeyConstraint(['operator_id'], ['users.id']),
sa.ForeignKeyConstraint(['serial_device_id'], ['serial_devices.id']),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('transaction_no'),
)
# 盘点记录表
op.create_table(
'inventory_checks',
sa.Column('id', sa.BigInteger(), autoincrement=True, nullable=False),
sa.Column('check_no', sa.String(length=50), nullable=False),
sa.Column('check_date', sa.Date(), nullable=False),
sa.Column('checker_id', sa.BigInteger(), nullable=True),
sa.Column('material_id', sa.BigInteger(), nullable=True),
sa.Column('batch_id', sa.BigInteger(), nullable=True),
sa.Column('book_quantity', sa.Integer(), nullable=True),
sa.Column('actual_quantity', sa.Integer(), nullable=True),
sa.Column('difference', sa.Integer(), nullable=True),
sa.Column('reason', sa.Text(), nullable=True),
sa.Column('adjusted', sa.Boolean(), server_default='false', nullable=True),
sa.Column('created_at', postgresql.TIMESTAMP(), server_default=sa.text('now()'), nullable=True),
sa.ForeignKeyConstraint(['batch_id'], ['inventory_batches.id']),
sa.ForeignKeyConstraint(['checker_id'], ['users.id']),
sa.ForeignKeyConstraint(['material_id'], ['materials.id']),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('check_no'),
)
# 新增库存管理权限
permissions_table = sa.table(
'permissions',
sa.column('name', sa.String),
sa.column('code', sa.String),
sa.column('module', sa.String),
sa.column('description', sa.Text),
)
op.bulk_insert(permissions_table, [
{'name': '查看库存', 'code': 'inventory.view', 'module': 'inventory', 'description': '查看库存列表和统计'},
{'name': '管理物料', 'code': 'inventory.manage', 'module': 'inventory', 'description': '创建、编辑、删除物料'},
{'name': '出入库操作', 'code': 'inventory.transaction', 'module': 'inventory', 'description': '执行采购入库、领用出库、退库操作'},
{'name': '库存盘点', 'code': 'inventory.check', 'module': 'inventory', 'description': '创建盘点单并执行库存调整'},
{'name': '库存报表', 'code': 'inventory.report', 'module': 'inventory', 'description': '查看库存统计报表'},
])
# 默认给 area_admin 查看权限
op.execute("""
INSERT INTO role_permissions (role, permission_id)
SELECT 'area_admin', id FROM permissions WHERE code = 'inventory.view'
""")
# 插入默认物料分类
op.execute("""
INSERT INTO material_categories (name, code, description) VALUES
('ONU设备', 'ONU', '光网络单元设备'),
('OLT设备', 'OLT', '光线路终端设备'),
('交换机', 'SWITCH', '网络交换机'),
('防火墙', 'FIREWALL', '网络防火墙设备'),
('上网行为管理', 'BEHAVIOR', '上网行为管理设备'),
('光模块及配件', 'ACCESSORY', '光模块、跳线等配件')
""")
def downgrade() -> None:
op.drop_table('inventory_checks')
op.drop_table('inventory_transactions')
op.drop_index('ix_serial_devices_status', table_name='serial_devices')
op.drop_index('ix_serial_devices_mac_address', table_name='serial_devices')
op.drop_table('serial_devices')
op.drop_table('inventory_batches')
op.drop_index('ix_materials_id', table_name='materials')
op.drop_table('materials')
op.drop_table('material_categories')
op.execute("""
DELETE FROM role_permissions WHERE permission_id IN (
SELECT id FROM permissions WHERE module = 'inventory'
)
""")
op.execute("DELETE FROM permissions WHERE module = 'inventory'")
Reference in New Issue View Git Blame Copy Permalink