Fastapi-mail

The fastapi-mail simple lightweight mail system, sending emails and attachments(individual && bulk)

🔨 Installation

 $ pip install fastapi-mail

In order to run the application use command below

uvicorn examples.main:app --reload --port 8001

Guide

fromfastapiimportFastAPI, BackgroundTasks, UploadFile, File, Formfromstarlette.responsesimportJSONResponsefromstarlette.requestsimportRequestfromfastapi_mailimportFastMail, MessageSchema,ConnectionConfigfrompydanticimportEmailStrfrompydanticimportEmailStr, BaseModelfromtypingimportListclassEmailSchema(BaseModel):
email: List[EmailStr]
conf=ConnectionConfig(
MAIL_USERNAME="YourUsername",
MAIL_PASSWORD="strong_password",
MAIL_FROM="your@email.com",
MAIL_PORT=587,
MAIL_SERVER="your mail server",
MAIL_TLS=True,
MAIL_SSL=False
)
app=FastAPI()
html="""<p>Hi this test mail, thanks for using Fastapi-mail</p> """template="""<p>Hi this test mail using BackgroundTasks, thanks for using Fastapi-mail</p> """@app.post("/email")asyncdefsimple_send(email: EmailSchema) ->JSONResponse:
message=MessageSchema(
subject="Fastapi-Mail module",
recipients=email.dict().get("email"), # List of recipients, as many as you can pass body=html,
subtype="html"
)
fm=FastMail(conf)
awaitfm.send_message(message)
returnJSONResponse(status_code=200, content={"message": "email has been sent"})

Sending email as background task

@app.post("/emailbackground")asyncdefsend_in_background(background_tasks: BackgroundTasks,email: EmailSchema) ->JSONResponse:
message=MessageSchema(
subject="Fastapi mail module",
recipients=email.dict().get("email"),
body="Simple background task ",
)
fm=FastMail(conf)
background_tasks.add_task(fm.send_message,message)
returnJSONResponse(status_code=200, content={"message": "email has been sent"})

Sending files

@app.post("/file")asyncdefsend_file(background_tasks: BackgroundTasks,file: UploadFile=File(...),email:EmailStr=Form(...)) ->JSONResponse:
message=MessageSchema(
subject="Fastapi mail module",
recipients=[email],
body="Simple background task ",
attachments=[file]
)
fm=FastMail(conf)
background_tasks.add_task(fm.send_message,message)
returnJSONResponse(status_code=200, content={"message": "email has been sent"})

Using Jinja2 HTML Templates

The email folder must be present within your applications working directory.

In sending HTML emails, the CSS expected by mail servers -outlook, google, etc- must be inline CSS. Fastapi mail passes "body" to the rendered template. In creating the template for emails the dynamic objects should be used with the assumption that the variable is named "body" and that it is a python dict.

check out jinja2 for more details https://jinja.palletsprojects.com/en/2.11.x/

classEmailSchema(BaseModel):
email: List[EmailStr]
body: Dict[str, Any]
conf=ConnectionConfig(
MAIL_USERNAME="YourUsername",
MAIL_PASSWORD="strong_password",
MAIL_FROM="your@email.com",
MAIL_PORT=587,
MAIL_SERVER="your mail server",
MAIL_TLS=True,
MAIL_SSL=False,
TEMPLATE_FOLDER='./email templates folder'
)
@app.post("/email")asyncdefsend_with_template(email: EmailSchema) ->JSONResponse:
message=MessageSchema(
subject="Fastapi-Mail module",
recipients=email.dict().get("email"), # List of recipients, as many as you can pass body=email.dict().get("body"),
subtype="html"
)
fm=FastMail(conf)
awaitfm.send_message(message, template_name="email_template.html") ##optional field template_name is the name of the html file(jinja template) to use from the email template folderreturnJSONResponse(status_code=200, content={"message": "email has been sent"})

Guide for email utils

The utility allows you to check temporary email addresses, you can block any email or domain. You can connect Redis to save and check email addresses. If you do not provide a Redis configuration, then the utility will save it in the list or set by default.

Check dispasoble email address

fromfastapiimportFastAPI, Query, Bodyfromstarlette.responsesimportJSONResponsefrompydanticimportEmailStrfromtypingimportListfromfastapi_mail.email_utilsimportDefaultCheckerfromfastapiimportDependsapp=FastAPI()
asyncdefdefault_checker():
checker=DefaultChecker() # you can pass source argument for your own email domainsawaitchecker.fetch_temp_email_domains() # require to fetch temporary email domainsreturnchecker@app.get('/email/dispasoble')asyncdefsimple_send(domain: str=Query(...), checker: DefaultChecker=Depends(default_checker)) ->JSONResponse:
ifawaitchecker.is_dispasoble(domain):
returnJSONResponse(status_code=400, content={'message': 'this is dispasoble domain'})
...
returnJSONResponse(status_code=200, content={'message': 'email has been sent'})

Add dispasoble email address

@app.post('/email/dispasoble')asyncdefadd_disp_domain(domains: list=Body(...,embed=True), checker: DefaultChecker=Depends(default_checker)) ->JSONResponse:
res=awaitchecker.add_temp_domain(domains)
returnJSONResponse(status_code=200, content={'result': res})

Add domain to blocked list

@app.post('/email/blocked/domains')asyncdefblock_domain(domain: str=Query(...), checker: DefaultChecker=Depends(default_checker)) ->JSONResponse:
awaitchecker.blacklist_add_domain(domain)
returnJSONResponse(status_code=200, content={'message': f'{domain} added to blacklist'})

Check domain blocked or not

@app.get('/email/blocked/domains')asyncdefget_blocked_domain(domain: str=Query(...), checker: DefaultChecker=Depends(default_checker)) ->JSONResponse:
res=awaitchecker.is_blocked_domain(domain)
returnJSONResponse(status_code=200, content={"result": res})

Add email address to blocked list

@app.post('/email/blocked/address')asyncdefblock_address(email: str=Query(...), checker: DefaultChecker=Depends(default_checker)) ->JSONResponse:
awaitchecker.blacklist_add_email(email)
returnJSONResponse(status_code=200, content={"result": True})

Check email blocked or not

@app.get('/email/blocked/address')asyncdefget_block_address(email: str=Query(...), checker: DefaultChecker=Depends(default_checker)) ->JSONResponse:
res=awaitchecker.is_blocked_address(email)
returnJSONResponse(status_code=200, content={"result": res})

Check MX record

@app.get('/email/mx')asyncdeftest_mx(email: EmailStr=Query(...),full_result: bool=Query(False) ,checker: DefaultChecker=Depends(default_checker)) ->JSONResponse:
domain=email.split("@")[-1]
res=awaitchecker.check_mx_record(domain,full_result)
returnJSONResponse(status_code=200, content=res)

Remove email address from blocked list

@app.delete('/email/blocked/address')asyncdefdel_blocked_address(email: str=Query(...), checker: DefaultChecker=Depends(default_checker)) ->JSONResponse:
res=awaitchecker.blacklist_rm_email(email)
returnJSONResponse(status_code=200, content={"result": res})

Remove domain from blocked list

@app.delete('/email/blocked/domains')asyncdefdel_blocked_domain(domain: str=Query(...), checker: DefaultChecker=Depends(default_checker)) ->JSONResponse:
res=awaitchecker.blacklist_rm_domain(domain)
returnJSONResponse(status_code=200, content={"result": res})

Remove domain from temporary list

@app.delete('/email/dispasoble')asyncdefdel_disp_domain(domains: list=Body(...,embed=True), checker: DefaultChecker=Depends(default_checker)) ->JSONResponse:
res=awaitchecker.blacklist_rm_temp(domains)
returnJSONResponse(status_code=200, content={'result': res})

Contributing

Fell free to open issue and send pull request.

Contributors ✨

Thanks goes to these wonderful people (🚧):

Sabuhi Shukurov 💬👀🚧	Tural Muradov 📖👀🔧	Hasan Aliyev 📖🚧👀	Ashwani 🚧	Leon Xu 🚧	Gabriel Oliveira 📖🚧	Onothoja Marho 📖🚧🔧
Tim Kiely 🚧

This project follows the all-contributors specification. Contributions of any kind are welcome!

Before you start please read CONTRIBUTING

LICENSE

MIT