pip install sqlalchemy

# 教程使用mysql数据库

pip install pymysql

from sqlalchemy import create_engine

import pymysql

pymysql.install_as_MySQLdb()

engine = create_engine("mysql://root:xxx@127.0.0.1:3306/company", echo=True)

from sqlalchemy import String, Integer

from sqlalchemy.orm import Mapped, mapped_column

from entity.base import Base

class Company(Base):

    __tablename__ = "company"

    code: Mapped[str] = mapped_column(String(32))

    name: Mapped[str] = mapped_column(String(32))

    exchange: Mapped[int] = mapped_column(Integer)

    data_status: Mapped[int] = mapped_column(Integer)

    disabled: Mapped[int] = mapped_column(Integer)

    def __repr__(self):

        return f'id:{self.id}, code:{self.code}, name:{self.name}, exchange:{self.exchange}, data_status:{self.data_status}, disabled:{self.disabled}, deleted:{self.deleted}'

from typing import Sequence

from integration.sql_alchemy import engine

from sqlalchemy import select, ScalarResult

from sqlalchemy.orm import Session

from entity.company import Company

class CompanyService:

    def __init__(self):

        pass

    def all(self) -> Sequence[Company]:

        with Session(engine) as session:

            stmt = select(Company).where(Company.disabled.__eq__(0)).where(Company.deleted.__eq__(0))

            result: ScalarResult[Company] = session.scalars(stmt)

            return result.all()

    def list_page(self, page: int, page_size: int, data_status: int = None) -> Sequence[Company]:

        with Session(engine) as session:

            stmt = select(Company).where(Company.disabled.__eq__(0)).where(Company.deleted.__eq__(0))

            if data_status is not None:

                stmt = stmt.where(Company.data_status.__eq__(data_status))

            stmt = stmt.limit(page_size).offset((page - 1) * page_size)

            result: ScalarResult[Company] = session.scalars(stmt)

            return result.all()

    def update_data_status(self, _id: int, data_status: int):

        with Session(engine) as session:

            stmt = select(Company).where(Company.id.__eq__(_id))

            company: Company = session.scalars(stmt).one()

            company.data_status = data_status

            session.commit()

from service.company_service import CompanyService

company_service = CompanyService()

company_list = company_service.list_page(1, 10)

for company in company_list:

    print(company)

create table company

(

    id          bigint unsigned auto_increment primary key,

    code        varchar(32)                not null,

    name        varchar(32)                not null,

    exchange    tinyint unsigned default 1 not null comment '1:上证,2:深证,3:创业板,4:科创板',

    data_status tinyint unsigned default 0 not null comment '0:未抓取,1:已抓取,2:已结构化',

    disabled    tinyint unsigned default 0 not null,

    deleted     tinyint unsigned default 0 not null,

    constraint code unique (code)

);