精华内容
下载资源
问答
  • 时间选择框具有readonly属性,在操作时不能直接使用send_keys输入值,需要使用js去掉readonly属性,然后在进行赋值 js = 'document.getElementById("inTimeDate").removeAttribute("readonly")' dr.execute_script...

    时间选择框具有readonly属性,在操作时不能直接使用send_keys输入值,需要使用js去掉readonly属性,然后在进行赋值

    js = 'document.getElementById("inTimeDate").removeAttribute("readonly")'
    dr.execute_script(js)
    dr.find_element_by_xpath(".//*[@id='inTimeDate']").send_keys(sign_in_time)

     

    展开全文
  • Python框架

    2019-08-19 13:32:22
    通过自动执行冗余任务,框架缩短了开发时间,使开发人员能够专注于应用程序逻辑而非常规元素。 因为它是领先的编程语言之一,所以Python的框架并不稀缺。不同的框架有各自的优点和问题。因此,需要根据项目要求和...

    原文地址: https://hackr.io/blog/python-frameworks

    Python程序员需要做些什么来让他们的生活更轻松?答案是框架。通过自动执行冗余任务,框架缩短了开发时间,使开发人员能够专注于应用程序逻辑而非常规元素。

    因为它是领先的编程语言之一,所以Python的框架并不稀缺。不同的框架有各自的优点和问题。因此,需要根据项目要求和开发者偏好进行选择。

    Python框架主要有三种类型,即全栈,微框架和异步。在继续讨论2019年最佳Python框架之前,让我们先简要介绍一下各种类型的Python框架。

    Python框架的类型

    1.全栈框架

    这些框架是所有开发人员要求的一站式解决方案。表单生成器,表单验证和模板布局通常可以使用典型的全栈框架。

    2.微框架

    这些是轻量级框架,不提供其他功能和特性,例如数据库抽象层,表单验证以及特定工具和库。使用微框架的开发人员需要手动添加大量代码和其他要求。

    3.异步框架

    最近获得普及,任何异步框架都是一个微框架,允许处理大量并发连接。通常,为Python构建的异步框架使用编程语言的asyncio库。

    2019年最佳Python框架

    AIOHTTP

    AIOhttp类型 - 异步框架

    AIOHTTP是一个Python框架,它在很大程度上依赖于Python 3.5+功能,例如async和awaits。Python框架使用Python的asyncio库,因此是一个异步框架。

    除了作为服务器Web框架之外,AIOHTTP还可以充当客户端框架。它提供了一个请求对象和路由器,可以将查询重定向到为处理相同问题而开发的函数。

    主要亮点:

    • 允许有效地构建视图
    • 中间件支持
    • 可插拔路由
    • 信号
    • 支持没有Callback Hell的Client WebSockets和Server WebSockets
    • 瓶子
    • 瓶子类型 - 微框架

    Bottle为使用它开发的每个应用程序创建单个源文件。Python的微框架最初是为构建API而开发的。除了Python标准库之外,Bottle没有制作小型Web应用程序所需的依赖项。

    使用Bottle的一个最重要的优点是它允许开发人员更接近硬件。除了构建简单的个人用途应用程序之外,Bottle还非常适合学习Web框架和原型设计的组织。

    主要亮点:

    • 适配器支持第三方模板引擎和WSGI / HTTP服务器
    • 允许从cookie,数据,文件上传和其他与HTTP相关的元数据进行简单访问
    • 内置HTTP服务器
    • 插件支持不同的数据库
    • 提供具有URL参数支持的请求分派路由

    CherryPy

    CherryPy的类型 - 微框架

    CherryPy是一种流行的开源,面向对象的Python框架,遵循简约的方法。微框架是最古老的Python框架之一,于2002年6月首次亮相。

    任何支持CherryPy的Web应用程序都是独立的Python应用程序,具有自己的嵌入式多线程Web服务器,可在任何支持Python的操作系统上运行。这样的应用程序可以部署在普通Python应用程序能够运行的任何地方。

    不需要Apache服务器来运行使用CherryPy开发的应用程序。微框架允许开发人员使用任何类型的技术进行数据访问,模板化等。

    主要亮点:

    • 许多用于身份验证,缓存,编码,会话,静态内容等的开箱即用工具
    • 灵活的内置插件系统
    • 符合HTTP / 1.1标准的WSGI线程池服务器
    • 内置的覆盖,分析和测试支持
    • 提供同时运行多个HTTP服务器的简便性
    • 强大的配置系统
    • 在Android上运行
    • CubicWeb
    • 立方网类型 - 全栈框架

    由Logilab开发和策划,CubicWeb是一个免费的,语义的,开源的,基于Python的Web框架。基于数据模型,CubicWeb需要具有相同的定义才能开发功能应用程序。

    与使用单独视图和模型的其他流行Python框架不同,CubicWeb使用cube。然后将多个多维数据集连接在一起,以便在数据库,Web服务器和一些配置文件的帮助下创建实例。

    主要亮点:

    • OWL(Web Ontology Language)和RDF(资源描述框架)支持
    • 可重复使用的组件
    • 安全工作流程
    • 使用RQL(关系查询语言)嵌入简化与数据相关的查询
    • 支持多个数据库
      短跑
      短跑类型 - 微框架

    Dash是一个基于Python的开源框架,用于构建分析Web应用程序。对于数据科学家来说,它是一个理想的Python框架,它不是Web开发的机制。

    使用Dash设计的应用程序是运行Flask并通过HTTP请求与JSON数据包通信的Web服务器。他们的前端使用ReactJS渲染组件。Flask插件可用于扩展Dash的功能。

    由于Dash应用程序在Web浏览器中呈现并可以部署到服务器,因此它们是跨平台和移动就绪的。Dash开发人员可以访问底层Flask实例及其所有可配置属性。

    主要亮点: -

    Dash应用程序需要很少的样板代码才能入门
    错误处理(Dash Deployment Server)
    高度定制化
    LDAP集成(Dash Deployment Server)
    插件支持
    用于绑定UI控件的简单界面,包括下拉列表,图形和滑块
    URL路由(Dash Deployment Server)
    Django的
    Django的类型 - 全栈框架

    Django是用于开发Python应用程序的最受欢迎的Web开发框架之一。事实上,它最终成为2019年十大网络开发框架之一。Django框架遵循DRY(d on’t ř EPEAT ý我们自己)的原则。

    与其他人不同,免费使用和开源的全栈Python框架包含大量内置功能,而不是将它们作为单独的库提供。Django利用其ORM将对象映射到数据库表。

    这样可以使代码跨不同的数据库工作,并使从一个数据库迁移到另一个数据库变得更容易。虽然Django对MySQL,PostgreSQL,SQLite和Oracle数据库有固有支持,但它可以通过第三方驱动程序支持其他数据库。

    主要亮点: -

    大量现成的库
    身份验证支持
    数据库架构迁移
    对象关系映射器(ORM)
    支持Web服务器
    模板引擎
    URL路由
    查看使用Django框架的优缺点。


    鹘类型 - 微框架

    为了快速构建Web API,Falcon是另一个广泛使用的Python框架。与需要加载大量依赖关系来构建HTTP API的其他Python框架不同,Falcon允许开发人员构建一个支持HTTP和REST架构的更简洁的设计。

    根据Sanic进行的基准测试,Falcon能够使用与同时代相同的硬件处理大多数请求。Python框架旨在实现100%的代码覆盖率。Falcon被LinkedIn,OpenStack和RackSpace等大公司使用。

    主要亮点: -

    可扩展,高度优化的代码库
    通过中间件组件和挂钩处理DRY请求
    通过请求和响应类轻松访问标题和正文
    Cython支持提高速度
    惯用的HTTP错误响应
    REST启发的资源类和URI模板提供直观的路由
    通过WSGI助手和模拟进行单元测试
    前期异常处理
    烧瓶
    烧瓶类型 - 微框架

    Flask是另一种流行的Python框架,在BSD许可下可用。灵感来自Sinatra Ruby框架,微框架需要Jinja2模板和Werkzeug WSGI工具包。由于其轻巧的模块化设计,Flask很容易适应。

    Flask允许开发人员构建一个可靠的Web应用程序基础,从而可以使用任何类型的扩展。微框架与Google App Engine兼容。

    主要亮点: -

    内置快速调试器
    HTTP请求处理
    内置开发服务器
    Jinja2模板
    RESTful请求调度
    支持插入任何ORM
    支持安全cookie以建立客户端会话
    基于Unicode
    单元测试支持
    WSGI 1.0合规性
    乔托
    乔托类型 - 全栈框架

    基于模型视图控制器模式,Giotto是Python的应用程序框架。为了让Web设计人员,Web开发人员和系统管理员能够独立工作,Giotto按顺序分离了Model,View和Controller元素。

    Giotto包含控制器模块,使用户能够在Web上创建应用程序,IRC(Internet中继聊天)和命令行。

    主要亮点: -

    自动URL路由
    使用SQLAlchemy进行数据库持久化
    非常简洁的代码
    功能性CRUD模式
    通用模型和视图
    内置缓存,支持Memcache和Redis(可用API,用于扩展对其他引擎的支持)
    用于HTML模板的Jinja2(可用于支持其他模板引擎的API)
    多个可插拔控制器
    RESTful界面与普通的“浏览器POST”CRUD站点
    咆哮者
    咆哮者类型 - 异步框架

    受NodeJS和Express / Connect框架的启发,Growler是一个在Python的asyncio库上编写的微Web框架。

    与其他传统的Python框架不同,Growler中的请求不是在框架中处理,而是通过中间件技术。

    作为易于快速实现复杂应用程序的Python框架的首选,Growler最初是由作者开发的,用于简单地学习如何在最低级别使用asyncio库。

    主要亮点: -

    由于缺少必要的回调和正确的try / except块,很容易看到程序流
    支持众多开源软件包
    使用装饰器编写干净,可重用的代码
    Ziapp模块允许将整个应用程序压缩为单个可执行文件
    拥抱
    类型 - 微框架

    拥抱
    Hug旨在允许Python开发人员开发一次API,然后在任何地方使用它。Python框架通过提供多个接口简化了API开发。它被标记为Python 3最快的Web框架。

    无论您是在进行本地开发还是通过HTTP或通过CLI,Hug都可以让您快速,轻松地完成应用程序开发。为了将性能提升到新的水平,Hug仅在需要时才消耗资源并使用Cython进行编译。

    主要亮点: -

    注释驱动的验证
    跟随写一次,随处使用哲学
    内置版本管理
    支持自动文档
    MorePath
    类型 - 微框架

    标签为“超级动力Python Web框架”,MorePath确保最小的设置占用空间。它专门用于使大多数典型用例尽快启动和运行,包括引入RESTful Web服务的常见Python数据结构。

    微框架,MorePath,是一个非常灵活的模型驱动的Web框架。Python框架可用的一个独特功能是,它可以自动将框架中定义的路径转换为使用最干净的URL创建应用程序的链接。

    主要亮点: -

    所有观点都是通用的。在视图中重用视图
    提供开发RESTful Web服务所需的所有工具
    创建通用用户界面就像子类化一样简单
    可扩展的简单,连贯,通用的扩展和覆盖机制
    灵活,简单,强大的权限
    Pycnic
    类型 - 微框架

    Pycnic是面向对象的微框架,被认为是构建基于JSON的API的最快速度。针对基于JSON的API框架的小型,独立和优化,能够在大型企业中保持良好的基础。由于Pycnic只创建Web API,因此占用空间极小,因此速度很快。

    主要亮点: -

    内置错误处理
    能够处理基于JSON的请求
    处理路由
    支持cookie
    Pylons框架
    主塔类型 - 全栈框架

    Pylons Framework是一个基于Python的开源框架,专注于应用程序的快速开发。该框架是通过结合一些最好的元素和语言属性(包括Perl,Python和Ruby)来设计的。

    尽管可以在维护模式下使用,但由于能够为Web开发提供高度灵活的结构,一些开发人员仍然使用Pylons框架。为了提高可重用性,全栈框架广泛使用WSGI(Web服务器网关接口)。

    主要亮点: -

    HTML表单验证和生成
    路线
    基于文本的模板
    网址发送
    基于WebHelper的路由配置的URL映射
    金字塔
    金字塔
    类型 - 全栈框架

    金字塔是另一个领先的Python框架。开源的基于Python的Web开发框架的主要目标是实现尽可能简单的复杂性。

    金字塔最令人向往的方面是能够同样适用于小型和全尺寸应用。此外,根据要求的要求,有许多资源可用于扩展其能力。

    主要亮点: -

    灵活的身份验证和授权
    功能装饰器
    谓词
    渲染器
    单文件应用程序
    模板和资产规格
    网址生成
    中信高科
    中信高科类型 - 异步框架

    Sanic构建于uvloop之上,是一个简单的开源Python框架,专门用于通过异步请求处理提供快速HTTP响应。因此,它是Python的异步框架。

    Sanic支持异步请求处理程序,使其与Python 3.5的async / await函数兼容。它可以提高速度并提供无阻塞功能。

    在一个进程和100个连接的基准测试期间,Sanic能够在一秒钟内处理多达33,342个请求。

    主要亮点: -

    能够读写cookie
    允许不同类型的日志记录,例如访问日志和错误日志
    基于类的视图
    易于应用装饰器的处理程序支持
    插件支持
    支持应用程序中的子路由蓝图
    可以使用点表示法或类似字典来修改配置对象
    龙卷风
    龙卷风类型 - 异步框架

    Tornado是一个开源Python框架和一个异步网络库。除了解决C10k问题(这意味着在任何给定时间处理10k连接),异步框架使用非阻塞网络I / O.

    Python框架最初是为一家名为FriendFeed的公司开发的,该公司于2009年被Facebook收购.Tornado是构建要求高性能和数千个并发用户的应用程序的理想工具。

    主要亮点: -

    允许实施第三方身份验证和授权方案
    提供高品质的输出
    实时服务
    支持翻译和本地化
    用户身份验证支持
    网页模板
    的TurboGears
    的TurboGears类型 - 全栈

    TurboGears是一个面向Python的数据驱动的开源,全栈Web应用程序框架。该框架允许开发人员快速开发可扩展的数据驱动的Web应用程序。

    除了支持灵活而强大的ORM之外,TurboGears还具有可理解的模板功能。全栈框架利用Genshi,Repoze,SQLAlchemy和WebOb等组件轻松快速地开发需要数据库连接的应用程序。

    主要亮点: -

    所有功能都实现为功能装饰器
    可用的命令行工具
    MochiKit JavaScript库集成
    多数据库支持
    MVC风格的架构
    PasteScript模板
    ToscaWidgets用于简化前端设计和服务器部署的协调
    使用Pylons作为Web服务器
    FormEncode的验证支持
    的web2py
    的web2py类型 - 全栈框架

    对于寻求可扩展的全栈框架的Python开发人员来说,Web2Py可能就是答案。开源Python框架附带了自己的基于Web的IDE,其中包括代码编辑器,调试器和一键式部署。

    虽然Web2Py允许用户在Python中创建动态Web内容,但它不支持Python 3.票务系统是Python框架最重要的功能之一。每当发生错误时,系统都会向用户发出票证。

    主要亮点: -

    能够在任何支持Python或Java和Python的Web托管平台上运行
    向后兼容性
    内置数据安全性,可防止多种常见漏洞,包括跨站点脚本,注入漏洞和恶意文件执行
    没有安装和配置要求
    遵循MVC模式
    为国际化提供支持
    多种协议的可读性
    基于角色的访问控制

    展开全文
  • Python 定时任务框架

    2021-03-05 10:20:19
    在实际开发中我们经常会碰上一些重复性或周期性的任务,比如像每天定时爬取某个网站的数据、一定周期定时运行代码训练模型等,类似这类的任务通常需要我们手动来进行设定或调度,以便其能够在我们设定好的时间内运行...

    一、APScheduler 简介

    在实际开发中我们经常会碰上一些重复性或周期性的任务,比如像每天定时爬取某个网站的数据、一定周期定时运行代码训练模型等,类似这类的任务通常需要我们手动来进行设定或调度,以便其能够在我们设定好的时间内运行。

    在 Windows 上我们可以通过计划任务来手动实现,而在 Linux 系统上往往我们会用到更多关于 crontab 的相关操作。但手动管理并不是一个很好的选择,如果我们需要有十几个不同的定时任务需要管理,那么每次通过人工来进行干预未免有些笨拙,那这时候就真的是「人工智能」了。

    所以将这些定时任务的调度代码化才是能够让我们很好地从这种手动管理的纯人力操作中解脱出来。

    在 Python 生态中对于定时任务的一些操作主要有那么几个:

    1. schedule:第三方模块,该模块适合比较轻量级的一些调度任务,但却不适用于复杂时间的调度

    2. APScheduler:第三方定时任务框架,是对 Java 第三方定时任务框架Quartz 的模仿与移植,能提供比 schedule 更复杂的应用场景,并且各种组件都是模块化,易于使用与二次开发。

    3. Celery Beat:属于 celery 这分布式任务队列第三方库下的一个定时任务组件,如果使用需要配合 RabbitMQ 或 Redis 这类的消息队列套件,需要花费一定的时间在环境搭建上,但在高版本中已经不支持 Windows。

    所以为了满足能够相对复杂的时间条件,又不需要在前期的环境搭建上花费很多时间的前提下,选择 APScheduler 来对我们的调度任务或定时任务进行管理是个性价比极高的选择。而本文主要会带你快速上手有关 APScheduler 的使用。

    二、APScheduler 概念与组件

    虽然说官方文档上的内容不是很多,而且所列举的 API 不是很多,但这侧面也反映了这一框架的简单易用。所以在使用 APScheduler 之前,我们需要对这个框架的一些概念简单了解,主要有那么以下几个:

    • 触发器(trigger)

    • 任务持久化(job stores)

    • 执行器(executor)

    • 调度器(scheduler)

    1.触发器(trigger)

    所谓的触发器就是用以触发定时任务的组件,在 APScheduler 中主要是指时间触发器,并且主要有三类时间触发器可供使用:

    • date:日期触发器。日期触发器主要是在某一日期时间点上运行任务时调用,是 APScheduler 里面最简单的一种触发器。所以通常也适用于一次性的任务或作业调度。

    • interval:间隔触发器。间隔触发器是在日期触发器基础上扩展了对时间部分,比如时、分、秒、天、周这几个部分的设定。是我们用以对重复性任务进行设定或调度的一个常用调度器。设定了时间部分之后,从起始日期开始(默认是当前)会按照设定的时间去执行任务。

    • croncron 表达式触发器。cron 表达式触发器就等价于我们 Linux 上的 crontab,它主要用于更复杂的日期时间进行设定。但需要注意的是,APScheduler 不支持 6 位及以上的 cron 表达式,最多只支持到 5 位。

    2.任务持久化(job stores)

    任务持久化主要是用于将设定好的调度任务进行存储,即便是程序因为意外情况,如断电、电脑或服务器重启时,只要重新运行程序时,APScheduler 就会根据对存储好的调度任务结果进行判断,如果出现已经过期但未执行的情况会进行相应的操作。

    APScheduler 为我们提供了多种持久化任务的途径,默认是使用 memory 也就是内存的形式,但内存并不是持久化最好的方式。最好的方式则是通过像数据库这样的载体来将我们的定时任务写入到磁盘当中,只要磁盘没有损坏就能将数据给恢复。

    APScheduler 支持的且常用的数据库主要有:

    • sqlalchemy 形式的数据库,这里就主要是指各种传统的关系型数据库,如 MySQL、PostgreSQL、SQLite 等。

    • mongodb 非结构化的 Mongodb 数据库,该类型数据库经常用于对非结构化或版结构化数据的存储或操作,如 JSON。

    • redis 内存数据库,通常用作数据缓存来使用,当然通过一些主从复制等方式也能实现当中数据的持久化或保存。

    通常我们可以在创建 Scheduler 实例时创建,或是单独为任务指定。配置的方式相对简单,我们只需要指定对应的数据库链接即可。

    3.执行器(executor)

    执行器顾名思义就是执行我们任务的对象,在计算机内通常要么是 CPU 调度任务,要么是单独维护一个线程来运行任务。所以 APScheduler 里的执行器通常就是 ThreadPoolExecutor 或 ProcessPoolExecutor 这样的线程池和进程池两种。

    当然如果是和协程或异步相关的任务调度,还可以使用对应的 AsyncIOExecutorTwistedExecutor 和 GeventExecutor 三种执行器。

    4.调度器(scheduler)

    调度器的选择主要取决于你当前的程序环境以及 APScheduler 的用途。根据用途的不同,APScheduler 又提供了以下几种调度器:

    • BlockingScheduler:阻塞调度器,当程序中没有任何存在主进程之中运行东西时,就则使用该调度器。

    • BackgroundScheduler:后台调度器,在不使用后面任何的调度器且希望在应用程序内部运行时的后台启动时才进行使用,如当前你已经开启了一个 Django 或 Flask 服务。

    • AsyncIOSchedulerAsyncIO 调度器,如果代码是通过 asyncio 模块进行异步操作,使用该调度器。

    • GeventSchedulerGevent 调度器,如果代码是通过 gevent 模块进行协程操作,使用该调度器

    • TornadoSchedulerTornado 调度器,在 Tornado 框架中使用

    • TwistedSchedulerTwisted 调度器,在基于 Twisted 的框架或应用程序中使用

    • QtSchedulerQt 调度器,在构建 Qt 应用中进行使用。

    通常情况下如果不是和 Web 项目或应用集成共存,那么往往都首选 BlockingScheduler 调度器来进行操作,它会在当前进程中启动相应的线程来进行任务调度与处理;反之,如果是和 Web 项目或应用共存,那么需要选择 BackgroundScheduler 调度器,因为它不会干扰当前应用的线程或进程状况。

    基于对以上的概念和组件认识,我们就能基本上摸清 APScheduler 的运行流程:

    1. 设定调度器(scheduler)用以对任务的调度与安排进行全局统筹

    2. 对相应的函数或方法上设定相应的触发器(trigger),并添加到调度器中

    3. 如有任务持久化(job stores)需要则需要设定对应的持久化层,否则默认使用内存存储任务

    4. 当触发器被触发时,就将任务交由执行器(executor)进行执行

     

    三、APScheduler 快速上手

    虽然 APScheduler 里面的概念和组件看起来有点多,但在使用上并不算很复杂,我们可以通过本节的示例就能够很快使用。

    1.选择对应的 scheduler

    在使用之前我们需要先实例化一个 scheduler 对象,所有的 scheduler 对象都被放在了 apscheduler.schedulers 模块下,我们可以直接通过查看 API 文档或者借助 IDE 补全的提示来获取相应的 scheduler 对象。

    这里我直接选取了最基础的 BlockingScheduler

    # main.py
    
    from apscheduler.schedulers.blocking import BlockingScheduler
    
    scheduler = BlockingScheduler()
    

    2.配置 scheduler

    对于 scheduler 的一些配置我们可以直接在实例化对象时就进行配置,当然也可以在创建实例化对象之后再进行配置。

    实例化时进行参数配置:

    # main.py
    from datetime import datetime
    
    from apscheduler.executors.pool import ThreadPoolExecutor
    from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
    from apscheduler.schedulers.blocking import BlockingScheduler
    
    # 任务持久化 使用 SQLite
    jobstores = {
        'default': SQLAlchemyJobStore(url = 'sqlite:///jobs.db')
    }
    # 执行器配置
    executors = {
        'default': ThreadPoolExecutor(20),
    }
    # 关于 Job 的相关配置,见官方文档 API
    job_defaults = {
        'coalesce': False,
        'next_run_time': datetime.now()
    }
    scheduler = BlockingScheduler(
      jobstores = jobstores,
      executors = executors,
      job_defaults = job_defaults,
      timezone = 'Asia/Shanghai'
    )
    

    或是通过 scheduler.configure 方法进行同样的操作:

    scheduler = BlockingScheduler()
    scheduler.configure(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone='Asia/Shanghai')
    

    3.添加并执行你的任务

    创建 scheduler 对象之后,我们需要调用其下的 add_job() 或是 scheduled_job() 方法来将我们需要执行的函数进行注册。前者是以传参的形式指定对应的函数名,而后者则是以装饰器的形式直接对我们要执行的函数进行修饰。

    比如我现在有一个输出此时此刻时间的函数 now()

    from datetime import datetime
    
    def now(trigger):
        print(f"trigger:{trigger} -> {datetime.now()}")
    

    然后我打算每 5 秒的时候运行一次,那我们使用 add_job() 可以这样写:

    if __name__ == '__main__':
        scheduler.add_job(now, trigger = "interval", args = ("interval",), seconds = 5)
        scheduler.start()
    

    在调用 start() 方法之后调度器就会开始执行,并在控制台上看到对应的结果了:

    trigger:interval -> 2021-01-16 21:19:43.356674
    trigger:interval -> 2021-01-16 21:19:46.679849
    trigger:interval -> 2021-01-16 21:19:48.356595
    

    当然使用 @scheduled_job 的方式来装饰我们的任务或许会更加自由一些,于是上面的例子就可以写成这样:

    @scheduler.scheduled_job(trigger = "interval", args = ("interval",), seconds = 5)
    def now(trigger):
        print(f"trigger:{trigger} -> {datetime.now()}")
    
    if __name__ == '__main__':
        scheduler.start()
    

    运行之后就会在控制台看到同样的结果了。

    不过需要注意的是,添加任务一定要在 start() 方法执行前调用,否则会找不到任务或是抛出异常。

    四、将 APScheduler 集成到 Web 项目中

    如果你是正在做有关的 Web 项目且存在一些定时任务,那么得益于APScheduler 由于多样的调度器,我们能够将其和我们的项目结合到一起。

    如果你正在使用 Flask,那么 Flask-APScheduler 这一别人写好的第三方包装库就很适合你,虽然它没有相关的文档,但只要你了解了前面我所介绍的有关于 APScheduler 的概念和组件,你就能很轻易地看懂这个第三方库仓库里的示例代码。

    如果你使用的不是 Flask 框架,那么 APScheduler 本身也提供了一些对任务或作业的增删改查操作,我们可以自己编写一套合适的 API。

    这里我使用的是 FastAPI 这一目前流行的 Web 框架。demo 项目结构如下:

    temp-scheduler
    ├── config.py       # 配置项
    ├── main.py         # API 文件
    └── scheduler.py    # APScheduler 相关设置
    

    1.安装依赖

    这里我们需要的依赖不多,只需要简单几个即可:

    pip install fastapi apscheduler sqlalchemy uvicorn
    

    2.配置项

    如果项目中模块过多,那么使用一个文件或模块来进行统一管理是最好的选择。这里的 config.py 我们主要像 Flask 的配置那样简单设定:

    from apscheduler.executors.pool import ThreadPoolExecutor
    from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
    from apscheduler.schedulers.blocking import BlockingScheduler
    
    class SchedulerConfig:
    
        JOBSTORES = {"default": SQLAlchemyJobStore(url="sqlite:///job.db")}
        EXECUTORS = {"default": ThreadPoolExecutor(20)}
        JOB_DEFAULTS = {"coalesce": False}
    
        @classmethod
        def to_dict(cls):
            return {
                "jobstores": cls.JOBSTORES,
                "executors": cls.EXECUTORS,
                "job_defaults": cls.JOB_DEFAULTS,
            }
    

    在 SchedulerConfig 配置项中我们可以自己实现一个 to_dict() 类方法,以便我们后续传参时通过解包的方式直接传入配置参数即可。

    3.Scheduler 相关设置

    scheduler.py 模块的设定也比较简单,即设定对应的 scheduler 调度器即可。由于是演示 demo 我还将要定期执行的任务也放在了这个模块当中:

    import logging
    from datetime import datetime
    
    from apscheduler.schedulers.background import BackgroundScheduler
    
    from config import SchedulerConfig
    
    scheduler = BackgroundScheduler()
    logger = logging.getLogger(__name__)
    
    def init_scheduler() -> None:
        # config scheduler
        scheduler.configure(**SchedulerConfig.to_dict())
    
        logger.info("scheduler is running...")
    
        # schedule test
        scheduler.add_job(
            func=mytask,
            trigger="date",
            args=("APScheduler Initialize.",),
            next_run_time=datetime.now(),
        )
        scheduler.start()
    
    def mytask(message: str) -> None:
        print(f"[{datetime.now()}] message: {message}")
    

    在这一部分中:

    • init_scheduler() 方法主要用于在 API 服务启动时被调用,然后对 scheduler 对象的配置以及测试

    • mytask() 则是我们要定期执行的任务,后续我们可以通过 APScheduler 提供的方法来自行添加任务

    4.API 设置

    在 main.py 模块就主要存放着我们由 FastAPI 所构建的相关 API。如果在后续开发时存在多个接口,此时就需要将不同接口放在不同模块文件中,以达到路由的分发与管理,类似于 Flask 的蓝图模式。

    import logging
    import uuid
    from datetime import datetime
    from typing import Any, Dict, Optional, Sequence, Union
    
    from fastapi import FastAPI
    from pydantic import BaseModel
    
    from scheduler import init_scheduler, mytask, scheduler
    
    logger = logging.getLogger(__name__)
    
    app = FastAPI(title="APScheduler API")
    app.add_event_handler("startup", init_scheduler)
    
    class Job(BaseModel):
        id: Union[int, str, uuid.UUID]
        name: Optional[str] = None
        func: Optional[str] = None
        args: Optional[Sequence[Optional[str]]] = None
        kwargs: Optional[Dict[str, Any]] = None
        executor: Optional[str] = None
        misfire_grace_time: Optional[str] = None
        coalesce: Optional[bool] = None
        max_instances: Optional[int] = None
        next_run_time: Optional[Union[str, datetime]] = None
    
    @app.post("/add")
    def add_job(
        message: str,
        trigger: str,
        trigger_args: Optional[dict],
        id: Union[str, int, uuid.UUID],
    ):
        try:
            scheduler.add_job(
                func=mytask,
                trigger=trigger,
                kwargs={"message": message},
                id=id,
                **trigger_args,
            )
        except Exception as e:
            logger.exception(e.args)
            return {"status_code": 0, "message": "添加失败"}
        return {"status_code": 1, "message": "添加成功"}
    
    @app.delete("/delete/{id}")
    def delete_job(id: Union[str, int, uuid.UUID]):
        """delete exist job by id"""
        try:
            scheduler.remove_job(job_id=id)
        except Exception:
            return dict(
                message="删除失败",
                status_code=0,
            )
        return dict(
            message="删除成功",
            status_code=1,
        )
    
    @app.put("/reschedule/{id}")
    def reschedule_job(
        id: Union[str, int, uuid.UUID], trigger: str, trigger_args: Optional[dict]
    ):
        try:
            scheduler.reschedule_job(job_id=id, trigger=trigger, **trigger_args)
        except Exception as e:
            logger.exception(e.args)
            return dict(
                message="修改失败",
                status_code=0,
            )
        return dict(
            message="修改成功",
            status_code=1,
        )
    
    @app.get("/job")
    def get_all_jobs():
        jobs = None
        try:
            job_list = scheduler.get_jobs()
            if job_list:
                jobs = [Job(**task.__getstate__()) for task in job_list]
        except Exception as e:
            logger.exception(e.args)
            return dict(
                message="查询失败",
                status_code=0,
                jobs=jobs,
            )
        return dict(
            message="查询成功",
            status_code=1,
            jobs=jobs,
        )
    
    @app.get("/job/{id}")
    def get_job_by_id(id: Union[int, str, uuid.UUID]):
        jobs = []
        try:
            job = scheduler.get_job(job_id=id)
            if job:
                jobs = [Job(**job.__getstate__())]
        except Exception as e:
            logger.exception(e.args)
            return dict(
                message="查询失败",
                status_code=0,
                jobs=jobs,
            )
        return dict(
            message="查询成功",
            status_code=1,
            jobs=jobs,
        )
    

    以上代码看起来很多,其实核心的就那么几点:

    1. FastAPI 对象 app 的初始化。这里用到的 add_event_handler() 方法就有点像 Flask 中的 before_first_request,会在 Web 服务请求伊始进行操作,理解为初始化相关的操作即可。

    2. API 接口路由。路由通过 app 对象下的对应 HTTP 方法来实现,如 GETPOSTPUT 等。这里的装饰器用法其实也和 Flask 很类似,就不多赘述。

    3. scheduler 对象的增删改查。从 scheduler.py 模块中引入我们创建好的 scheduler 对象之后就可以直接用来做增删改查的操作:

      1. 增:使用 add_job() 方法,其主要的参数是要运行的函数(或方法)、触发器以及触发器参数等

      2. 删:使用 delete_job() 方法,我们需要传入一个对应任务的 id 参数,用以能够查找到对应的任务

      3. 改:使用 reschedule_job() 方法,这里也需要一个对应任务的 id 参数,以及需要重新修改的触发器及其参数

      4. 查:使用 get_jobs() 和 get_job() 两个方法,前者是直接获取到当前调度的所有任务,返回的是一个包含了 APScheduler.job.Job 对象的列表,而后者是通过 id 参数来查找对应的任务对象;这里我通过底层源码使用 __getstate__() 来获取到任务的相关信息,这些信息我们通过事先设定好的 Job 对象来对其进行序列化,最后将信息从接口中返回。

    5.运行

    完成以上的所有操作之后,我们就可以打开控制台,进入到该目录下并激活我们的虚拟环境,之后运行:

    uvicorn main:app 
    

    之后我们就能在 FastAPI 默认的地址 http://127.0.0.1:8000/docs  中看到关于全部接口的 Swagger 文档页面了:

    图片

    fastapi 集成的 swagger 页面

    之后我们可以直接在文档里面或使用 Postman 来自己进行接口测试即可。

    五、结尾

    本文介绍了有关于 APScheduler 框架的概念及其用法,并进行了简单的实践。

    得益于 APScheduler 的模块化设计才可以让我们更方便地去理解、使用它,并将其运用到我们实际的开发过程中。

    从 APScheduler 目前的 Github 仓库代码以及 issue 来看,作者已经在开始重构 4.0 版本,当中的一些源代码和 API 也有较大的变动,相信在 4.0 版本中将会引入更多的新特性。

    展开全文
  • 再newproject中选择Django工程,在Location中输入建立的项目文件夹名称,Application name中输入要建立的app名称后电机Create。 2.在templates文件夹中创建想要使用的html文件。 html文件内容: <!DOCTYPE...

    1.创建Django工程

    再newproject中选择Django工程,在Location中输入建立的项目文件夹名称,Application name中输入要建立的app名称后电机Create。

    2.在templates文件夹中创建想要使用的html文件。

    html文件内容:

    <!DOCTYPE html>
    <html lang="en">
    <head>
        <meta charset="UTF-8">
        <title>Title</title>
    </head>
    <body>
        <h1>time: {{ time }}</h1>
    </body>
    </html>
    

      {{ }}为特殊标记,标记出前端中等待通过后端传入值的参数。

    3.在blog文件夹下的views.py文件下加入如下程序:

    from django.shortcuts import render
    import time
    
    # Create your views here.
    
    def show_time(req):
        t = time.ctime()
        return render(req, 'index.html', {'time': t})
    

      view是视图的意思,自定义函数传入参数是访问的请求信息,返回值是服务器的响应信息

      在render(着色)函数中第一个参数是请求参数,第二个参数是需要渲染的html文件,第三个参数是字典其键对应html中的变量,其值对应服务器中的变量。

    4.在Django的文件夹下的urls.py文件中加入如下程序:

    """Django URL Configuration
    
    The `urlpatterns` list routes URLs to views. For more information please see:
        https://docs.djangoproject.com/en/2.1/topics/http/urls/
    Examples:
    Function views
        1. Add an import:  from my_app import views
        2. Add a URL to urlpatterns:  path('', views.home, name='home')
    Class-based views
        1. Add an import:  from other_app.views import Home
        2. Add a URL to urlpatterns:  path('', Home.as_view(), name='home')
    Including another URLconf
        1. Import the include() function: from django.urls import include, path
        2. Add a URL to urlpatterns:  path('blog/', include('blog.urls'))
    """
    from django.contrib import admin
    from django.urls import path
    from blog import views
    
    urlpatterns = [
        path('admin/', admin.site.urls),
        path('show_time/', views.show_time),
    ]
    

      urlpatterns函数是服务器请求的分发函数,path中的第一个参数是通过正则函数查询的内容,通过GET拿到的字符串中含有的部分,找到对应的函数。

    5.在Terminal窗口中输入以下指令启动框架运行。

    E:\Django>python manage.py runserver 127.0.0.1:8081

    成功效果如下:

    Performing system checks...
    
    System check identified no issues (0 silenced).
    
    You have 15 unapplied migration(s). Your project may not work properly until you apply the migrations for app(s): admin, auth, contenttypes,
    sessions.
    Run 'python manage.py migrate' to apply them.
    December 21, 2018 - 21:20:07
    Django version 2.1.4, using settings 'Django.settings'
    Starting development server at http://127.0.0.1:8081/
    Quit the server with CTRL-BREAK.
    

       manage.py是这个Djando工程的启动函数,输入IP地址和监听端口号可以开始监听。

    6.在网页中访问http://127.0.0.1:8081/show_time/ 得到访问结果:

    通过以上6步完成通过Django框架显示时间功能。

     

    转载于:https://www.cnblogs.com/CK85/p/10159159.html

    展开全文
  • 好几年前,做web开发时,我们选择Python的tornado框架,看重的是它的轻量级(简洁)和异步(说实话,没怎么用异步)。 后来,开始用Python 3,同时也接触了asyncio,对异步IO有了跟深入的理解和掌握。asyncio模块...
  • 这周写selenium+python编写自动化用例遇到的难点 首先是安装好selenium和python,不赘述 难点1:关于上传图片 自动化测试上传图片,借用了autoIt这个工具,编写脚本,然后把脚本转换成.exe,在代码中再调用这个 .exe...
  • web后台框架 1. Django 、Flask、Tornado的对比 答案: Django走的是大而全的方向,开发...重量级web框架,功能齐全,提供一站式解决的思路,能让开发者不用在选择上花费大量时间。 自带ORM(Object-Relational Mapp...
  • 然而现在大多数web程序往往不是富内容的, 这似乎让Django看起来不是最佳选择的web框架。 那么让我们花点时间从现在的web开发实践中重新认识下她吧。 简单清晰的Django 一个web框架主要是帮助web程序生成核心的架构...
  • 估计好多朋友Linux中的crontab定时任务并不陌生,其实python中也是有定时任务框架的,接下来一起看看吧。... 基于间隔的执行(以偶数间隔运行作业,也可以选择开始/结束时间) 一次性延迟执行
  • 学习Python也有段时间了,一直在写自动化测试和小工具,准备通过搭建个人网站再全面深入整理下相关知识。选择Flask是基于两个特点:一、轻便,二、扩展性强。 准备工作 环境 系统:Mac big sur 开发工具:Visual ...
  • 一般情况下,使用Flask框架开发者大多数都是选择Python虚拟环境来运行项目,不同的虚拟环境中配置依赖包信息不同。如果重新迁移到一个新的虚拟环境后,又重新来一个一个的配置依赖包,那将会很浪费时间。 下面介绍一...
  • 好久以来都一直想学习windows中得iocp技术,即异步通信,但是经过长时间研究别人的c++版本,发现过于深奥了,有点吃力,不过幸好python中的twisted技术的存在方便了我。  iocp即异步通信技术,是windows系统中现在...
  • 技术环境: PyCharm + Flask + Python3.7 + Redis + mysql 功能说明: 1.用户在没有登录的时候点击买东西,会弹出首先登录,同时也可以选择注册,注册需要邮箱验证码 2.商品有二级分类,可以按照大类或小类列表展示...
  • 打算用python写点gui程序, 之前听说过python的gui库有 wxPython, pyQt, 查Google后, 又有两个候选项目, pygtk, Tkinter. 选项真够多的, 有时候选项太多, 也不是什么好事. 花了我大把时间搜索和试用. 选项:1. pyQt ...
  • 通过使用非阻塞网络I / O,Tornado可以扩展到数万个开放连接,使其成为需要长时间连接每个用户的长轮询,是WebSockets和其他应用程序的理想选择。 Tornado 和现在的主流 Web 服务器框架(包括大多数 Python 的框架...
  • 在实际开发中我们经常会碰上一些重复性或周期性的任务,比如像每天定时爬取某个网站的数据、一定周期定时运行代码训练模型等,类似这类的任务通常需要我们手动来进行设定或调度,以便其能够在我们设定好的时间内运行...
  • 由于Python具有快速开发,多人协作,库丰富,社区成熟等优点,国内豆瓣/知乎/搜狐/果壳/饿了吗,国外YouTube/Reddit/Instagram/Pinterest/Bitbucket/Disqus/Dropbox等都相继将其作为主力开发语言,所以完全不用担心它的...
  • Python 最快Web框架

    2021-04-09 10:43:15
    点上方“菜鸟学Python”,选择“星标” 463篇原创干货,第一时间送达 作者:ConnorZhang链接:https://juejin.cn/post/6944598601674784...
  • 这个是unittest框架的运行方式,有时不是很便利,比如:生成测试报告时,文件时间命名无法正常添加。 解决:  这时候我们可以修改。... 这里指定以python 的方式运行,Script内选择需要运行的模块。 ...
  • 姜斯图 Django框架实现学生信息管理系统 [视频链接]!( ) ...可以选择多条记录进行删除操作,还可以在课程列表页面可以对不同课程进行排序。 数据页码: 可以设置相应数据在每一页中显示的数量多少,
  • 一般情况下,使用Flask框架开发者大多数都是选择Python虚拟环境来运行项目,不同的虚拟环境中配置依赖包信息不同。如果重新迁移到一个新的虚拟环境后,又重新来一个一个的配置依赖包,那将会很浪费时间。 下面...
  • 本次个人作业我选择的语言是Python,了解学习Python有一段时间了但是一直没有练习,所以这次玩蛇,使用的版本是Python3.6。 开发工具选择选择的IDE是Pycharm,个人认为Pycharm是一款不错的Py开发工具,用起来...
  • 由于最近项目组要做爬虫需要一些自动化的工作,一直没时间,今天在此做个记录,给需要的小伙伴~ 在我们的日常工作自动化测试当中,几乎超过一半的功能都需要利用定时的任务来推动触发,例如在我们项目中有一个定时...
  • 提到Python的web框架,大部分人肯定第一时间想到的是Flask、Django等,这些框架可谓是大名鼎鼎,但本文讲述的是最近比较火的高性能框架FastApi。 为什么选择FastApi? 它是目前Python最快的web框架之一,根据官网...
  • 点击“Python编程与实战”,选择“置顶公众号”第一时间获取 Python 技术干货!ASGI在 Python3.5 之后增加 async/await 特性之后,异步编程变得异常火爆,...
  • 使用win7的Python3.73的IDLE界面编写图形界面程序,在编码过程中发现,如果定义一个主窗口后再定义一个局部窗口会导致后面的窗口的编辑设置和取值异常。具体问题用一个测试程序来说明如下: ``` import ...
  • 2019独角兽企业重金招聘Python工程师标准>>> ...

空空如也

空空如也

1 2 3 4 5 ... 20
收藏数 543
精华内容 217
关键字:

python时间选择框

python 订阅