搜索到
453
篇与
的结果
-
FastApi操作MySQL基本文件结构 首先是总的文件结构：.├── __pycache__│ └── main.cpython-38.pyc├── main.py└── sqlstu ├── __pycache__ │ ├── crud.cpython-38.pyc │ ├── database.cpython-38.pyc │ └── models.cpython-38.pyc ├── crud.py ├── database.py └── models.pymain.py就是FastAPI的启动文件了。 main.py:import uvicorn from pydantic import BaseModel from sqlstu import crud from sqlalchemy.orm import Session from fastapi import Response, Depends, FastAPI, HTTPException from sqlstu.database import SessionLocal, engine from sqlstu import models app = FastAPI() models.Base.metadata.create_all(bind=engine) # 数据库开关 def get_db(): db = SessionLocal() try: yield db finally: db.close() class Item(BaseModel): form: str form_name: str entry: dict @app.get('/') async def index(db: Session = Depends(get_db)): a = '111' b = '222' crud.add_record(db, a, b) return 1 if __name__ == '__main__': uvicorn.run(app='main:app', host='0.0.0.0', port=8000, reload=True, debug=False)crud.py进行数据库增删改查from sqlalchemy.orm import Session from .models import * def add_record(db: Session, source, result): if not db.query(PandaParse).filter(PandaParse.source == source).first(): added = PandaParse(source=source, result=result) db.add(added) db.commit() db.refresh(added) return 200database.py建立数据库链接from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL = "mysql+pymysql://username:password@localhost:port/database" engine = create_engine(SQLALCHEMY_DATABASE_URL, pool_pre_ping=True) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()models.py数据库表对应文件from sqlalchemy import Boolean, Column, Integer, String, Datetime from datetime import datetime from .database import Base class PandaParse(Base): __tablename__ = "LanAPI_Panda" id = Column(Integer, primary_key=True, index=True) source = Column(String) result = Column(String) -
-
Linux定时任务，Django_crontab设置，Django定时任务设置 Django_crontab仅限于Linux系统（唉，可惜呀），这是我经过试验的出来的结论。首先呢，需要安装一下django_crontab，一如既往的pippip install django_crontab然后在Django的settings中添加app，似乎要添置到前排才行。INSTALLED_APPS = [ 'django_crontab', ...... ]再在settings中添加CRONJOBS = [ ('*/1 * * * *', 'Wei.core.task', ">> result.log") ]这里的CRONJOBS固定的写法不用管，中括号内便是自己的任务。首先是第一个参数：'*/1 * * * *'这个呢，和Linux下的crontab的执行时间计算一样的。例子： # 每月的最后1天 0 0 L * * * 说明： Linux * * * * * - - - - - | | | | | | | | | +----- day of week (0 - 7) (Sunday=0 or 7) OR sun,mon,tue,wed,thu,fri,sat | | | +---------- month (1 - 12) OR jan,feb,mar,apr ... | | +--------------- day of month (1 - 31) | +-------------------- hour (0 - 23) +------------------------- minute (0 - 59)字段是否必填允许值允许特殊字符备注Seconds是0–59*,-标准实现不支持此字段。Minutes是0–59*,-Hours是0–23*,-Day of month是1–31*,-?LW?LW只有部分软件实现了Month是1–12 or JAN–DEC*,-Day of week是0–7 or SUN–SAT*,-?L#?L#只有部分软件实现了Linux和Spring的允许值为0-7，0和7为周日Quartz的允许值为1-7，1为周日Year否1970–2099*,-标准实现不支持此字段。这个可以用网上的一些功能进行在线计算，百度自寻即可。第二个参数，指定任务所在'Wei.core.task'我的这个意思就是在Wei这个app下的core.py中的task函数第三个参数，指定日志文件">> result.log"签名的>>表示写入日志时追加写入，如需覆盖则为一个>后面的就是文件路径了
-
-
-
sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread. 爬学校电费余额系统存到本地数据库时，开了多线程（车速过快），于是就报错了sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread. The object was created in thread id 12960 and this is thread id 13904.在链接sqlite的时候加上check_same_thread=False即可
-
